FN-6954: reconcile stale parked task assignments
Reconcile agent/task drift when durable agents remain linked to queued tasks without live execution proof. - clear stale Agent.taskId links for parked todo/triage tasks while preserving task leases and queue state - report stale parked assignments as active/no-live-run in Reports Health Check before reconciliation completes - add scheduler and self-healing coverage for queued lease drift, overlap starvation, and audit events - document the reconciliation behavior and add a published package patch changeset Files changed: .changeset/fn-6954-agent-task-state-drift.md | 5 + docs/architecture.md | 2 + .../src/__tests__/heartbeat-executor.test.ts | 68 ++++++++++++ .../__tests__/scheduler-overlap-starvation.test.ts | 68 +++++++++++- .../self-healing-agent-link-drift.test.ts | 97 ++++++++++++++++- .../engine/src/__tests__/task-agent-sync.test.ts | 33 +++++- packages/engine/src/agent-heartbeat.ts | 121 +++++++++++++++++++-- packages/engine/src/run-audit.ts | 5 + packages/engine/src/runtimes/in-process-runtime.ts | 1 + packages/engine/src/scheduler.ts | 24 +++- packages/engine/src/self-healing.ts | 102 ++++++++++++++--- packages/engine/src/task-agent-sync.ts | 65 ++++++++++- 12 files changed, 555 insertions(+), 36 deletions(-) Fusion-Task-Id: FN-6954 Fusion-Task-Lineage: 24b8a2eb-5a33-4539-ab64-ae2bbfc2d195
This commit is contained in:
5
.changeset/fn-6954-agent-task-state-drift.md
Normal file
5
.changeset/fn-6954-agent-task-state-drift.md
Normal file
@@ -0,0 +1,5 @@
|
|||||||
|
---
|
||||||
|
"@runfusion/fusion": patch
|
||||||
|
---
|
||||||
|
|
||||||
|
Fix stale durable agent task assignments for tasks parked behind file-scope lease queues, including Reports Health Check rendering and self-healing reconciliation.
|
||||||
@@ -682,6 +682,7 @@ Runtime action-gate flow (v1):
|
|||||||
- `recoverGhostReviewTasks()` is a fallback only for idle, non-terminal `in-review` states. Terminal/actionable states (notably `status: "failed"`) are preserved and **not** auto-kicked back to `todo`.
|
- `recoverGhostReviewTasks()` is a fallback only for idle, non-terminal `in-review` states. Terminal/actionable states (notably `status: "failed"`) are preserved and **not** auto-kicked back to `todo`.
|
||||||
- `recoverPausedAbortFailures()` clears executor pause/resume abort parks only when the durable row is safe to recover. `todo`/`in-progress` rows are requeued for normal scheduling, while clean `in-review` rows (completed steps, not paused/user-paused/executing, auto-merge eligible, no confirmed or terminal merge evidence) have `status`/`error` cleared in place so review progression can continue. User hard-cancel, global/user pause, `autoMerge:false`, terminal merge, and live-execution guards remain operator-actionable. Successful recovery emits `task:auto-recover-paused-abort-park` with `preservedInReview` metadata.
|
- `recoverPausedAbortFailures()` clears executor pause/resume abort parks only when the durable row is safe to recover. `todo`/`in-progress` rows are requeued for normal scheduling, while clean `in-review` rows (completed steps, not paused/user-paused/executing, auto-merge eligible, no confirmed or terminal merge evidence) have `status`/`error` cleared in place so review progression can continue. User hard-cancel, global/user pause, `autoMerge:false`, terminal merge, and live-execution guards remain operator-actionable. Successful recovery emits `task:auto-recover-paused-abort-park` with `preservedInReview` metadata.
|
||||||
- `reattach-orphaned-assigned-executions` is a forward-resume safety net for durable-agent assignments. During startup recovery and periodic maintenance, after orphaned-agent and stale-heartbeat-run repairs, self-healing finds `in-progress` tasks with an `assignedAgentId` whose agent has no active heartbeat run and no active executor session after the orphan grace window. It re-dispatches in place via `executor.resumeTaskForAgent(agentId)` (the same seam used by clean `HeartbeatMonitor.onRunCompleted` and guarded by executor double-execution checks), emits `task:reattach-orphaned-execution`, and never moves the task backward. This complements engine-start `executor.resumeOrphaned()` and leaves unassigned/role-based execution recovery to the existing startup/limbo/stuck-task paths.
|
- `reattach-orphaned-assigned-executions` is a forward-resume safety net for durable-agent assignments. During startup recovery and periodic maintenance, after orphaned-agent and stale-heartbeat-run repairs, self-healing finds `in-progress` tasks with an `assignedAgentId` whose agent has no active heartbeat run and no active executor session after the orphan grace window. It re-dispatches in place via `executor.resumeTaskForAgent(agentId)` (the same seam used by clean `HeartbeatMonitor.onRunCompleted` and guarded by executor double-execution checks), emits `task:reattach-orphaned-execution`, and never moves the task backward. This complements engine-start `executor.resumeOrphaned()` and leaves unassigned/role-based execution recovery to the existing startup/limbo/stuck-task paths.
|
||||||
|
- Durable `Agent.taskId` is a running assignment for parked `todo`/`triage` task rows only when the agent has live proof: a fresh active heartbeat run or an executor-active/tracked heartbeat signal. Scheduler overlap requeues, task move sync, self-healing, and Reports Health Check share this invariant: stale durable links are cleared or rendered as stale while `status: "queued"` and `overlapBlockedBy` remain on the task row so file-scope lease blocking is not weakened.
|
||||||
- Mission validation has a dedicated stale-run reaper: startup recovery and Batch 2 maintenance call `reapStaleMissionValidatorRuns()` when wired by the runtime, using `VALIDATOR_RUN_STALE_MAX_AGE_MS` (currently 6 hours). The sweep terminates ownerless `mission_validator_runs.status='running'` rows as `error`, writes the reap reason into `summary`, leaves `lastValidatorRunId` pointing at the now-terminal run, and emits run-audit telemetry with `mutationType: "mission:validator-run-reaped"` plus `runId`/`featureId`/`missionId`/`triggerType`/`elapsedMs` metadata. Active mission features move to `loopState="needs_fix"` + `lastValidatorStatus="error"` unless their parent mission is already `complete`/`archived`.
|
- Mission validation has a dedicated stale-run reaper: startup recovery and Batch 2 maintenance call `reapStaleMissionValidatorRuns()` when wired by the runtime, using `VALIDATOR_RUN_STALE_MAX_AGE_MS` (currently 6 hours). The sweep terminates ownerless `mission_validator_runs.status='running'` rows as `error`, writes the reap reason into `summary`, leaves `lastValidatorRunId` pointing at the now-terminal run, and emits run-audit telemetry with `mutationType: "mission:validator-run-reaped"` plus `runId`/`featureId`/`missionId`/`triggerType`/`elapsedMs` metadata. Active mission features move to `loopState="needs_fix"` + `lastValidatorStatus="error"` unless their parent mission is already `complete`/`archived`.
|
||||||
|
|
||||||
#### Stuck-loop exhaustion terminal contract
|
#### Stuck-loop exhaustion terminal contract
|
||||||
@@ -1091,6 +1092,7 @@ The run-audit system records every mutation performed by the engine across four
|
|||||||
- **Database / `task:no-commits-finalize-blocked-incomplete-steps`** — emitted by no-op finalize lanes when a `noCommitsExpected` task has no net branch changes but incomplete/skipped steps outweigh done steps. Metadata includes `{ reason, doneCount, incompleteCount, lane, classification?, baseRef? }`; the accompanying task log explains that the task was demoted to `todo` with progress preserved instead of finalized as done.
|
- **Database / `task:no-commits-finalize-blocked-incomplete-steps`** — emitted by no-op finalize lanes when a `noCommitsExpected` task has no net branch changes but incomplete/skipped steps outweigh done steps. Metadata includes `{ reason, doneCount, incompleteCount, lane, classification?, baseRef? }`; the accompanying task log explains that the task was demoted to `todo` with progress preserved instead of finalized as done.
|
||||||
- **Database / `task:orphan-detected-no-action`** — emitted by `recoverOrphanedExecutions` (FN-5337) when row metadata looks orphaned after grace windows; annotation-only event with no lifecycle mutation (`in-progress` task stays put).
|
- **Database / `task:orphan-detected-no-action`** — emitted by `recoverOrphanedExecutions` (FN-5337) when row metadata looks orphaned after grace windows; annotation-only event with no lifecycle mutation (`in-progress` task stays put).
|
||||||
- **Database / `task:reattach-orphaned-execution`** — emitted by `reattachOrphanedAssignedExecutions` (FN-6336) when self-healing re-dispatches an idle assigned `in-progress` task forward via `executor.resumeTaskForAgent(agentId)` after proving the assigned agent has no active heartbeat run or active execution.
|
- **Database / `task:reattach-orphaned-execution`** — emitted by `reattachOrphanedAssignedExecutions` (FN-6336) when self-healing re-dispatches an idle assigned `in-progress` task forward via `executor.resumeTaskForAgent(agentId)` after proving the assigned agent has no active heartbeat run or active execution.
|
||||||
|
- **Database / `task:reconcile-stale-agent-assignment`** — emitted when self-healing or heartbeat reconciliation clears stale durable `Agent.taskId`/`state` for a task parked in `todo`/`triage` without live execution proof. Metadata includes `{ agentId, taskId, taskColumn, agentState, status, blockedBy, overlapBlockedBy, hadFreshRun, hadActiveExecution, reason }`; task queue/lease fields are preserved.
|
||||||
- **Database / `task:soft-delete-column-reconciled`** — emitted by `reconcileSoftDeletedColumnDrift` (FN-5566, re-land FN-5446) when a soft-deleted row (`deletedAt IS NOT NULL`) is found with legacy `column != 'archived'`; rewrites only `column` (no resurrection), with metadata `{ previousColumn }`.
|
- **Database / `task:soft-delete-column-reconciled`** — emitted by `reconcileSoftDeletedColumnDrift` (FN-5566, re-land FN-5446) when a soft-deleted row (`deletedAt IS NOT NULL`) is found with legacy `column != 'archived'`; rewrites only `column` (no resurrection), with metadata `{ previousColumn }`.
|
||||||
- **Database / `session:runtime-resolved`** — emitted once per `createResolvedAgentSession` call with metadata `{ sessionPurpose, runtimeId, wasConfigured, provider, modelId, mockProviderActive, testModeActive, runtimeHint? }` for per-lane runtime/provider attribution.
|
- **Database / `session:runtime-resolved`** — emitted once per `createResolvedAgentSession` call with metadata `{ sessionPurpose, runtimeId, wasConfigured, provider, modelId, mockProviderActive, testModeActive, runtimeHint? }` for per-lane runtime/provider attribution.
|
||||||
- **Database / `task:reconcile-dependency-blocking-lease`** — emitted by `reconcileDependencyBlockingLeases()` (FN-6292) when self-healing rebounds an `in-progress` holder to `todo` because an unmet dependency is blocked by the holder's stale file-scope lease. Metadata includes the dependency ID, blocked-by marker, and unmet dependency list.
|
- **Database / `task:reconcile-dependency-blocking-lease`** — emitted by `reconcileDependencyBlockingLeases()` (FN-6292) when self-healing rebounds an `in-progress` holder to `todo` because an unmet dependency is blocked by the holder's stale file-scope lease. Metadata includes the dependency ID, blocked-by marker, and unmet dependency list.
|
||||||
|
|||||||
@@ -190,6 +190,8 @@ describe("executeHeartbeat", () => {
|
|||||||
setLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
setLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
||||||
clearLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
clearLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
||||||
appendRunLog: vi.fn().mockResolvedValue(undefined),
|
appendRunLog: vi.fn().mockResolvedValue(undefined),
|
||||||
|
getActiveHeartbeatRun: vi.fn().mockResolvedValue(null),
|
||||||
|
syncExecutionTaskLink: vi.fn().mockResolvedValue(undefined),
|
||||||
getAgentsByReportsTo: vi.fn().mockResolvedValue([]),
|
getAgentsByReportsTo: vi.fn().mockResolvedValue([]),
|
||||||
} as unknown as AgentStore;
|
} as unknown as AgentStore;
|
||||||
}
|
}
|
||||||
@@ -236,6 +238,72 @@ describe("executeHeartbeat", () => {
|
|||||||
expect(section).toContain("healthy");
|
expect(section).toContain("healthy");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("FN-6954: buildReportsHealthSection suppresses running state for parked task with no live proof", async () => {
|
||||||
|
const now = new Date().toISOString();
|
||||||
|
const store = createStoreWithAgentForExec();
|
||||||
|
vi.mocked(store.getAgentsByReportsTo).mockResolvedValue([
|
||||||
|
{ id: "agent-backend", name: "Backend Engineer", state: "running", taskId: "FN-6709", lastHeartbeatAt: now, updatedAt: now } as Agent,
|
||||||
|
]);
|
||||||
|
vi.mocked(store.getActiveHeartbeatRun).mockResolvedValue(null);
|
||||||
|
mockTaskStore = createMockTaskStore({
|
||||||
|
getTask: vi.fn(async (taskId: string) => ({
|
||||||
|
id: taskId,
|
||||||
|
column: "todo",
|
||||||
|
status: "queued",
|
||||||
|
overlapBlockedBy: "FN-6827",
|
||||||
|
blockedBy: null,
|
||||||
|
dependencies: [],
|
||||||
|
log: [],
|
||||||
|
steps: [],
|
||||||
|
attachments: [],
|
||||||
|
createdAt: now,
|
||||||
|
updatedAt: now,
|
||||||
|
}) as unknown as TaskDetail),
|
||||||
|
});
|
||||||
|
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||||
|
|
||||||
|
const section = await (monitor as any).buildReportsHealthSection("agent-001", store);
|
||||||
|
|
||||||
|
expect(section).toContain("| Backend Engineer | active | FN-6709 (queued/no live run) |");
|
||||||
|
expect(section).toContain("**stale** assignment");
|
||||||
|
expect(section).not.toContain("| Backend Engineer | running | FN-6709 |");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("FN-6954: buildReportsHealthSection preserves running state for parked task with fresh active run", async () => {
|
||||||
|
const now = new Date().toISOString();
|
||||||
|
const store = createStoreWithAgentForExec();
|
||||||
|
vi.mocked(store.getAgentsByReportsTo).mockResolvedValue([
|
||||||
|
{ id: "agent-backend", name: "Backend Engineer", state: "running", taskId: "FN-6709", lastHeartbeatAt: now, updatedAt: now } as Agent,
|
||||||
|
]);
|
||||||
|
vi.mocked(store.getActiveHeartbeatRun).mockResolvedValue({
|
||||||
|
id: "run-live",
|
||||||
|
agentId: "agent-backend",
|
||||||
|
startedAt: now,
|
||||||
|
status: "active",
|
||||||
|
} as AgentHeartbeatRun);
|
||||||
|
mockTaskStore = createMockTaskStore({
|
||||||
|
getTask: vi.fn(async (taskId: string) => ({
|
||||||
|
id: taskId,
|
||||||
|
column: "todo",
|
||||||
|
status: "queued",
|
||||||
|
overlapBlockedBy: "FN-6827",
|
||||||
|
dependencies: [],
|
||||||
|
log: [],
|
||||||
|
steps: [],
|
||||||
|
attachments: [],
|
||||||
|
createdAt: now,
|
||||||
|
updatedAt: now,
|
||||||
|
}) as unknown as TaskDetail),
|
||||||
|
});
|
||||||
|
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||||
|
|
||||||
|
const section = await (monitor as any).buildReportsHealthSection("agent-001", store);
|
||||||
|
|
||||||
|
expect(section).toContain("| Backend Engineer | running | FN-6709 |");
|
||||||
|
expect(section).not.toContain("queued/no live run");
|
||||||
|
expect(section).not.toContain("**stale** assignment");
|
||||||
|
});
|
||||||
|
|
||||||
it("buildReportsHealthSection classifies stuck agents", async () => {
|
it("buildReportsHealthSection classifies stuck agents", async () => {
|
||||||
const now = Date.now();
|
const now = Date.now();
|
||||||
const store = createStoreWithAgentForExec();
|
const store = createStoreWithAgentForExec();
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
import { beforeEach, describe, expect, it, vi } from "vitest";
|
import { beforeEach, describe, expect, it, vi } from "vitest";
|
||||||
import { Scheduler } from "../scheduler.js";
|
import { Scheduler } from "../scheduler.js";
|
||||||
import type { Task, TaskStore } from "@fusion/core";
|
import type { Agent, AgentStore, Task, TaskStore } from "@fusion/core";
|
||||||
|
|
||||||
function makeTask(overrides: Partial<Task> = {}): Task {
|
function makeTask(overrides: Partial<Task> = {}): Task {
|
||||||
return {
|
return {
|
||||||
@@ -18,6 +18,23 @@ function makeTask(overrides: Partial<Task> = {}): Task {
|
|||||||
} as Task;
|
} as Task;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
function createAgentStore(agents: Agent[]): AgentStore {
|
||||||
|
return {
|
||||||
|
listAgents: vi.fn(async (filter?: { state?: Agent["state"]; includeEphemeral?: boolean }) => {
|
||||||
|
return agents.filter((agent) => !filter?.state || agent.state === filter.state);
|
||||||
|
}),
|
||||||
|
getActiveHeartbeatRun: vi.fn(async () => null),
|
||||||
|
updateAgentState: vi.fn(async (agentId: string, state: Agent["state"]) => {
|
||||||
|
const agent = agents.find((candidate) => candidate.id === agentId);
|
||||||
|
if (agent) agent.state = state;
|
||||||
|
}),
|
||||||
|
syncExecutionTaskLink: vi.fn(async (agentId: string, taskId?: string) => {
|
||||||
|
const agent = agents.find((candidate) => candidate.id === agentId);
|
||||||
|
if (agent) agent.taskId = taskId;
|
||||||
|
}),
|
||||||
|
} as unknown as AgentStore;
|
||||||
|
}
|
||||||
|
|
||||||
function createStore(tasks: Task[], scopes: Record<string, string[]>): TaskStore {
|
function createStore(tasks: Task[], scopes: Record<string, string[]>): TaskStore {
|
||||||
const updateTask = vi.fn(async (id: string, patch: Partial<Task>) => {
|
const updateTask = vi.fn(async (id: string, patch: Partial<Task>) => {
|
||||||
const task = tasks.find((candidate) => candidate.id === id);
|
const task = tasks.find((candidate) => candidate.id === id);
|
||||||
@@ -149,6 +166,55 @@ describe("scheduler overlap starvation regression (FN-057)", () => {
|
|||||||
expect(store.moveTask).toHaveBeenCalledWith("FN-031", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
|
expect(store.moveTask).toHaveBeenCalledWith("FN-031", "in-progress", expect.objectContaining({ allocateWorktree: expect.any(Function) }));
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("FN-6954: clears stale running durable agents when overlap requeue parks todo task", async () => {
|
||||||
|
const tasks = [
|
||||||
|
makeTask({ id: "FN-6827", column: "in-progress", priority: "normal" }),
|
||||||
|
makeTask({ id: "FN-6709", column: "todo", priority: "urgent" }),
|
||||||
|
];
|
||||||
|
const agents = [{ id: "agent-backend", state: "running", taskId: "FN-6709" } as Agent];
|
||||||
|
const agentStore = createAgentStore(agents);
|
||||||
|
const store = createStore(tasks, {
|
||||||
|
"FN-6827": ["packages/engine/src/scheduler.ts"],
|
||||||
|
"FN-6709": ["packages/engine/src/scheduler.ts"],
|
||||||
|
});
|
||||||
|
|
||||||
|
const scheduler = new Scheduler(store, { agentStore, hasActiveAgentExecution: () => false });
|
||||||
|
(scheduler as any).running = true;
|
||||||
|
await scheduler.schedule();
|
||||||
|
|
||||||
|
expect(store.updateTask).toHaveBeenCalledWith("FN-6709", {
|
||||||
|
status: "queued",
|
||||||
|
blockedBy: null,
|
||||||
|
overlapBlockedBy: "FN-6827",
|
||||||
|
});
|
||||||
|
expect(agents[0]).toMatchObject({ state: "active", taskId: undefined });
|
||||||
|
expect((agentStore as any).updateAgentState).toHaveBeenCalledWith("agent-backend", "active");
|
||||||
|
expect((agentStore as any).syncExecutionTaskLink).toHaveBeenCalledWith("agent-backend", undefined);
|
||||||
|
expect(tasks.find((task) => task.id === "FN-6709")).toMatchObject({ status: "queued", overlapBlockedBy: "FN-6827" });
|
||||||
|
});
|
||||||
|
|
||||||
|
it("FN-6954: preserves running durable agent when overlap requeue has live execution proof", async () => {
|
||||||
|
const tasks = [
|
||||||
|
makeTask({ id: "FN-6827", column: "in-progress", priority: "normal" }),
|
||||||
|
makeTask({ id: "FN-6709", column: "todo", priority: "urgent" }),
|
||||||
|
];
|
||||||
|
const agents = [{ id: "agent-backend", state: "running", taskId: "FN-6709" } as Agent];
|
||||||
|
const agentStore = createAgentStore(agents);
|
||||||
|
const store = createStore(tasks, {
|
||||||
|
"FN-6827": ["packages/engine/src/scheduler.ts"],
|
||||||
|
"FN-6709": ["packages/engine/src/scheduler.ts"],
|
||||||
|
});
|
||||||
|
|
||||||
|
const scheduler = new Scheduler(store, { agentStore, hasActiveAgentExecution: (agentId) => agentId === "agent-backend" });
|
||||||
|
(scheduler as any).running = true;
|
||||||
|
await scheduler.schedule();
|
||||||
|
|
||||||
|
expect((agentStore as any).updateAgentState).not.toHaveBeenCalled();
|
||||||
|
expect((agentStore as any).syncExecutionTaskLink).not.toHaveBeenCalled();
|
||||||
|
expect(agents[0]).toMatchObject({ state: "running", taskId: "FN-6709" });
|
||||||
|
expect(tasks.find((task) => task.id === "FN-6709")).toMatchObject({ status: "queued", overlapBlockedBy: "FN-6827" });
|
||||||
|
});
|
||||||
|
|
||||||
it("does not defer FN-078-style ready work behind non-runnable queued overlaps", async () => {
|
it("does not defer FN-078-style ready work behind non-runnable queued overlaps", async () => {
|
||||||
const tasks = [
|
const tasks = [
|
||||||
makeTask({ id: "FN-069", column: "todo", status: "queued", priority: "high" }),
|
makeTask({ id: "FN-069", column: "todo", status: "queued", priority: "high" }),
|
||||||
|
|||||||
@@ -12,6 +12,7 @@ describe("FN-4296: self-healing agent link drift", () => {
|
|||||||
function buildManager(agents: Agent[], tasks: Record<string, Task | null>, hasActiveAgentExecution?: (agentId: string) => boolean) {
|
function buildManager(agents: Agent[], tasks: Record<string, Task | null>, hasActiveAgentExecution?: (agentId: string) => boolean) {
|
||||||
const store = {
|
const store = {
|
||||||
getTask: vi.fn(async (taskId: string) => tasks[taskId] ?? null),
|
getTask: vi.fn(async (taskId: string) => tasks[taskId] ?? null),
|
||||||
|
recordRunAuditEvent: vi.fn(async () => {}),
|
||||||
} as any;
|
} as any;
|
||||||
|
|
||||||
const agentStore = {
|
const agentStore = {
|
||||||
@@ -22,6 +23,10 @@ describe("FN-4296: self-healing agent link drift", () => {
|
|||||||
return agents;
|
return agents;
|
||||||
}),
|
}),
|
||||||
getActiveHeartbeatRun: vi.fn(async () => null),
|
getActiveHeartbeatRun: vi.fn(async () => null),
|
||||||
|
updateAgentState: vi.fn(async (agentId: string, state: Agent["state"]) => {
|
||||||
|
const agent = agents.find((candidate) => candidate.id === agentId);
|
||||||
|
if (agent) agent.state = state;
|
||||||
|
}),
|
||||||
syncExecutionTaskLink: vi.fn(async (agentId: string, taskId?: string) => {
|
syncExecutionTaskLink: vi.fn(async (agentId: string, taskId?: string) => {
|
||||||
const agent = agents.find((candidate) => candidate.id === agentId);
|
const agent = agents.find((candidate) => candidate.id === agentId);
|
||||||
if (agent) agent.taskId = taskId;
|
if (agent) agent.taskId = taskId;
|
||||||
@@ -29,7 +34,7 @@ describe("FN-4296: self-healing agent link drift", () => {
|
|||||||
} as unknown as AgentStore;
|
} as unknown as AgentStore;
|
||||||
|
|
||||||
const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore, hasActiveAgentExecution });
|
const manager = new SelfHealingManager(store, { rootDir: "/tmp/test-project", agentStore, hasActiveAgentExecution });
|
||||||
return { manager, agentStore };
|
return { manager, agentStore, store };
|
||||||
}
|
}
|
||||||
|
|
||||||
it("FN-4296: durable agent linked to done task is cleared by sweep", async () => {
|
it("FN-4296: durable agent linked to done task is cleared by sweep", async () => {
|
||||||
@@ -56,6 +61,96 @@ describe("FN-4296: self-healing agent link drift", () => {
|
|||||||
manager.stop();
|
manager.stop();
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("FN-6954: running durable agent on dependency-only queued todo is made active and unlinked", async () => {
|
||||||
|
const agents = [makeAgent("agent-backend", "FN-7000", "running")];
|
||||||
|
const queuedTask = {
|
||||||
|
id: "FN-7000",
|
||||||
|
column: "todo",
|
||||||
|
status: "queued",
|
||||||
|
blockedBy: "FN-6999",
|
||||||
|
overlapBlockedBy: null,
|
||||||
|
} as Task;
|
||||||
|
const { manager } = buildManager(agents, { "FN-7000": queuedTask }, () => false);
|
||||||
|
|
||||||
|
await manager.recoverDriftedAgentTaskLinks();
|
||||||
|
|
||||||
|
expect(agents[0]).toMatchObject({ state: "active", taskId: undefined });
|
||||||
|
expect(queuedTask).toMatchObject({ status: "queued", blockedBy: "FN-6999", overlapBlockedBy: null });
|
||||||
|
manager.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("FN-6954: running durable agent on overlap-queued triage task is made active and unlinked", async () => {
|
||||||
|
const agents = [makeAgent("agent-backend", "FN-7001", "running")];
|
||||||
|
const queuedTask = {
|
||||||
|
id: "FN-7001",
|
||||||
|
column: "triage",
|
||||||
|
status: "queued",
|
||||||
|
overlapBlockedBy: "FN-6827",
|
||||||
|
} as Task;
|
||||||
|
const { manager } = buildManager(agents, { "FN-7001": queuedTask }, () => false);
|
||||||
|
|
||||||
|
await manager.recoverDriftedAgentTaskLinks();
|
||||||
|
|
||||||
|
expect(agents[0]).toMatchObject({ state: "active", taskId: undefined });
|
||||||
|
expect(queuedTask).toMatchObject({ status: "queued", overlapBlockedBy: "FN-6827" });
|
||||||
|
manager.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("FN-6954: duplicate durable agents linked to one parked task preserve only live proof", async () => {
|
||||||
|
const agents = [
|
||||||
|
makeAgent("agent-stale", "FN-7002", "running"),
|
||||||
|
makeAgent("agent-live", "FN-7002", "running"),
|
||||||
|
];
|
||||||
|
const queuedTask = { id: "FN-7002", column: "todo", status: "queued", overlapBlockedBy: "FN-6827" } as Task;
|
||||||
|
const { manager } = buildManager(agents, { "FN-7002": queuedTask }, (agentId) => agentId === "agent-live");
|
||||||
|
|
||||||
|
await manager.recoverDriftedAgentTaskLinks();
|
||||||
|
|
||||||
|
expect(agents[0]).toMatchObject({ state: "active", taskId: undefined });
|
||||||
|
expect(agents[1]).toMatchObject({ state: "running", taskId: "FN-7002" });
|
||||||
|
expect(queuedTask).toMatchObject({ status: "queued", overlapBlockedBy: "FN-6827" });
|
||||||
|
manager.stop();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("FN-6954: running durable agent on lease-queued todo is made active and audited without clearing the lease", async () => {
|
||||||
|
const agents = [makeAgent("agent-backend", "FN-6709", "running")];
|
||||||
|
const queuedTask = {
|
||||||
|
id: "FN-6709",
|
||||||
|
column: "todo",
|
||||||
|
status: "queued",
|
||||||
|
overlapBlockedBy: "FN-6827",
|
||||||
|
blockedBy: null,
|
||||||
|
} as Task;
|
||||||
|
const blockerTask = { id: "FN-6827", column: "in-progress", assignedAgentId: "agent-other" } as Task;
|
||||||
|
const { manager, agentStore, store } = buildManager(
|
||||||
|
agents,
|
||||||
|
{ "FN-6709": queuedTask, "FN-6827": blockerTask },
|
||||||
|
() => false,
|
||||||
|
);
|
||||||
|
|
||||||
|
await manager.recoverDriftedAgentTaskLinks();
|
||||||
|
|
||||||
|
expect(agents[0]).toMatchObject({ state: "active", taskId: undefined });
|
||||||
|
expect(queuedTask).toMatchObject({ status: "queued", overlapBlockedBy: "FN-6827" });
|
||||||
|
expect((agentStore as any).updateAgentState).toHaveBeenCalledWith("agent-backend", "active");
|
||||||
|
expect(store.recordRunAuditEvent).toHaveBeenCalledWith(expect.objectContaining({
|
||||||
|
mutationType: "task:reconcile-stale-agent-assignment",
|
||||||
|
target: "agent-backend",
|
||||||
|
metadata: expect.objectContaining({
|
||||||
|
agentId: "agent-backend",
|
||||||
|
taskId: "FN-6709",
|
||||||
|
taskColumn: "todo",
|
||||||
|
agentState: "running",
|
||||||
|
status: "queued",
|
||||||
|
overlapBlockedBy: "FN-6827",
|
||||||
|
hadFreshRun: false,
|
||||||
|
hadActiveExecution: false,
|
||||||
|
reason: expect.stringContaining("without fresh run or active execution"),
|
||||||
|
}),
|
||||||
|
}));
|
||||||
|
manager.stop();
|
||||||
|
});
|
||||||
|
|
||||||
it("FN-4296: durable agent linked to todo task with fresh active run is NOT cleared", async () => {
|
it("FN-4296: durable agent linked to todo task with fresh active run is NOT cleared", async () => {
|
||||||
const agents = [makeAgent("agent-1", "FN-1")];
|
const agents = [makeAgent("agent-1", "FN-1")];
|
||||||
const { manager, agentStore } = buildManager(agents, { "FN-1": { id: "FN-1", column: "todo" } as Task }, () => true);
|
const { manager, agentStore } = buildManager(agents, { "FN-1": { id: "FN-1", column: "todo" } as Task }, () => true);
|
||||||
|
|||||||
@@ -3,7 +3,7 @@ import { mkdtempSync, rmSync } from "node:fs";
|
|||||||
import { tmpdir } from "node:os";
|
import { tmpdir } from "node:os";
|
||||||
import { join } from "node:path";
|
import { join } from "node:path";
|
||||||
|
|
||||||
import { AgentStore, type AgentCreateInput, type Task } from "@fusion/core";
|
import { AgentStore, type Agent, type AgentCreateInput, type Task } from "@fusion/core";
|
||||||
import { describe, expect, it, vi } from "vitest";
|
import { describe, expect, it, vi } from "vitest";
|
||||||
|
|
||||||
import { attachAgentLinkSync } from "../task-agent-sync.js";
|
import { attachAgentLinkSync } from "../task-agent-sync.js";
|
||||||
@@ -20,11 +20,17 @@ class EventedStore extends EventEmitter {
|
|||||||
const createInput: AgentCreateInput = { name: "durable-agent", role: "executor" };
|
const createInput: AgentCreateInput = { name: "durable-agent", role: "executor" };
|
||||||
|
|
||||||
describe("FN-4296: task agent sync", () => {
|
describe("FN-4296: task agent sync", () => {
|
||||||
const runCase = async (to: string, hasActiveAgentExecution = false) => {
|
const runCase = async (to: string, hasActiveAgentExecution = false, agentState: Agent["state"] = "active") => {
|
||||||
const store = new EventedStore();
|
const store = new EventedStore();
|
||||||
|
const agents = [{ id: "agent-1", taskId: "FN-1", state: agentState }];
|
||||||
const agentStore = {
|
const agentStore = {
|
||||||
listAgents: vi.fn(async () => [{ id: "agent-1", taskId: "FN-1" }]),
|
listAgents: vi.fn(async () => agents),
|
||||||
syncExecutionTaskLink: vi.fn(async () => undefined),
|
updateAgentState: vi.fn(async (_agentId: string, state: Agent["state"]) => {
|
||||||
|
agents[0].state = state;
|
||||||
|
}),
|
||||||
|
syncExecutionTaskLink: vi.fn(async (_agentId: string, taskId?: string) => {
|
||||||
|
agents[0].taskId = taskId;
|
||||||
|
}),
|
||||||
assignTask: vi.fn(async () => undefined),
|
assignTask: vi.fn(async () => undefined),
|
||||||
} as any;
|
} as any;
|
||||||
|
|
||||||
@@ -39,7 +45,7 @@ describe("FN-4296: task agent sync", () => {
|
|||||||
await Promise.resolve();
|
await Promise.resolve();
|
||||||
await Promise.resolve();
|
await Promise.resolve();
|
||||||
|
|
||||||
return { detach, agentStore };
|
return { detach, agentStore, agents };
|
||||||
};
|
};
|
||||||
|
|
||||||
it("FN-4296: task:moved → done clears linked durable agent's taskId", async () => {
|
it("FN-4296: task:moved → done clears linked durable agent's taskId", async () => {
|
||||||
@@ -57,8 +63,16 @@ describe("FN-4296: task agent sync", () => {
|
|||||||
expect(agentStore.syncExecutionTaskLink).toHaveBeenCalledWith("agent-1", undefined);
|
expect(agentStore.syncExecutionTaskLink).toHaveBeenCalledWith("agent-1", undefined);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("FN-6954: task:moved in-progress → todo queued by overlap clears stale running state", async () => {
|
||||||
|
const { agentStore, agents } = await runCase("todo", false, "running");
|
||||||
|
expect(agentStore.updateAgentState).toHaveBeenCalledWith("agent-1", "active");
|
||||||
|
expect(agentStore.syncExecutionTaskLink).toHaveBeenCalledWith("agent-1", undefined);
|
||||||
|
expect(agents[0]).toMatchObject({ state: "active", taskId: undefined });
|
||||||
|
});
|
||||||
|
|
||||||
it("FN-4296: task:moved → todo does NOT clear link when hasActiveAgentExecution=true", async () => {
|
it("FN-4296: task:moved → todo does NOT clear link when hasActiveAgentExecution=true", async () => {
|
||||||
const { agentStore } = await runCase("todo", true);
|
const { agentStore } = await runCase("todo", true, "running");
|
||||||
|
expect(agentStore.updateAgentState).not.toHaveBeenCalled();
|
||||||
expect(agentStore.syncExecutionTaskLink).not.toHaveBeenCalled();
|
expect(agentStore.syncExecutionTaskLink).not.toHaveBeenCalled();
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -67,6 +81,13 @@ describe("FN-4296: task agent sync", () => {
|
|||||||
expect(agentStore.syncExecutionTaskLink).toHaveBeenCalledWith("agent-1", undefined);
|
expect(agentStore.syncExecutionTaskLink).toHaveBeenCalledWith("agent-1", undefined);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("FN-6954: task:moved → triage queued behind overlap clears stale running link", async () => {
|
||||||
|
const { agentStore, agents } = await runCase("triage", false, "running");
|
||||||
|
expect(agentStore.updateAgentState).toHaveBeenCalledWith("agent-1", "active");
|
||||||
|
expect(agentStore.syncExecutionTaskLink).toHaveBeenCalledWith("agent-1", undefined);
|
||||||
|
expect(agents[0]).toMatchObject({ state: "active", taskId: undefined });
|
||||||
|
});
|
||||||
|
|
||||||
it("FN-4296: task:moved → in-review does NOT clear link", async () => {
|
it("FN-4296: task:moved → in-review does NOT clear link", async () => {
|
||||||
const { agentStore } = await runCase("in-review", false);
|
const { agentStore } = await runCase("in-review", false);
|
||||||
expect(agentStore.syncExecutionTaskLink).not.toHaveBeenCalled();
|
expect(agentStore.syncExecutionTaskLink).not.toHaveBeenCalled();
|
||||||
|
|||||||
@@ -35,7 +35,7 @@ import { buildPromptLayers, collapsePromptLayers } from "./prompt-layers.js";
|
|||||||
import { resolveAndEmitGoalContext } from "./goal-injection-diagnostics.js";
|
import { resolveAndEmitGoalContext } from "./goal-injection-diagnostics.js";
|
||||||
import { createLogger, heartbeatLog, formatError } from "./logger.js";
|
import { createLogger, heartbeatLog, formatError } from "./logger.js";
|
||||||
import { acquireTaskWorktree } from "./worktree-acquisition.js";
|
import { acquireTaskWorktree } from "./worktree-acquisition.js";
|
||||||
import { createRunAuditor, type EngineRunContext } from "./run-audit.js";
|
import { createRunAuditor, generateSyntheticRunId, type DatabaseMutationType, type EngineRunContext } from "./run-audit.js";
|
||||||
import { promptWithFallback } from "./pi.js";
|
import { promptWithFallback } from "./pi.js";
|
||||||
import { createResolvedAgentSession, extractRuntimeHint, resolveHeartbeatSessionModels } from "./agent-session-helpers.js";
|
import { createResolvedAgentSession, extractRuntimeHint, resolveHeartbeatSessionModels } from "./agent-session-helpers.js";
|
||||||
import type { AgentActionGateContext } from "./agent-action-gate.js";
|
import type { AgentActionGateContext } from "./agent-action-gate.js";
|
||||||
@@ -44,6 +44,7 @@ import type { AgentReflectionService } from "./agent-reflection.js";
|
|||||||
import { trimPromptMd, trimTaskDescription, trimTriggeringComments } from "./heartbeat-prompt-trim.js";
|
import { trimPromptMd, trimTaskDescription, trimTriggeringComments } from "./heartbeat-prompt-trim.js";
|
||||||
import { detectDeicticReference, extractAntecedentCandidates, renderAmbiguityPromptBlock, scoreReferentConfidence } from "./room-ambiguity.js";
|
import { detectDeicticReference, extractAntecedentCandidates, renderAmbiguityPromptBlock, scoreReferentConfidence } from "./room-ambiguity.js";
|
||||||
import { countActiveAgentMembers, decideRoomCoordination, detectTaskFilingIntent, renderRoomCoordinationPromptBlock } from "./room-coordination.js";
|
import { countActiveAgentMembers, decideRoomCoordination, detectTaskFilingIntent, renderRoomCoordinationPromptBlock } from "./room-coordination.js";
|
||||||
|
import { evaluateParkedAgentTaskLink, isParkedTaskColumn, type AgentTaskLinkExecutionProof } from "./task-agent-sync.js";
|
||||||
|
|
||||||
const promptSizeLog = createLogger("prompt-size");
|
const promptSizeLog = createLogger("prompt-size");
|
||||||
|
|
||||||
@@ -1256,6 +1257,45 @@ export class HeartbeatMonitor {
|
|||||||
}, this.pollIntervalMs);
|
}, this.pollIntervalMs);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private async emitStaleAgentAssignmentAudit(options: {
|
||||||
|
agent: Pick<Agent, "id" | "state">;
|
||||||
|
taskId: string;
|
||||||
|
linkedTask: TaskDetail | null;
|
||||||
|
hadFreshRun: boolean;
|
||||||
|
hadActiveExecution: boolean;
|
||||||
|
reason: string;
|
||||||
|
}): Promise<void> {
|
||||||
|
if (!this.taskStore) return;
|
||||||
|
try {
|
||||||
|
await createRunAuditor(this.taskStore, {
|
||||||
|
runId: generateSyntheticRunId("heartbeat-stale-agent-assignment", options.taskId),
|
||||||
|
agentId: "heartbeat-monitor",
|
||||||
|
taskId: options.taskId,
|
||||||
|
taskLineageId: options.linkedTask?.lineageId,
|
||||||
|
phase: "reconcile-stale-agent-assignment",
|
||||||
|
}).database({
|
||||||
|
type: "task:reconcile-stale-agent-assignment" as DatabaseMutationType,
|
||||||
|
target: options.agent.id,
|
||||||
|
metadata: {
|
||||||
|
agentId: options.agent.id,
|
||||||
|
taskId: options.taskId,
|
||||||
|
taskColumn: options.linkedTask?.column ?? null,
|
||||||
|
agentState: options.agent.state,
|
||||||
|
status: options.linkedTask?.status ?? null,
|
||||||
|
blockedBy: options.linkedTask?.blockedBy ?? null,
|
||||||
|
overlapBlockedBy: options.linkedTask?.overlapBlockedBy ?? null,
|
||||||
|
hadFreshRun: options.hadFreshRun,
|
||||||
|
hadActiveExecution: options.hadActiveExecution,
|
||||||
|
reason: options.reason,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
} catch (error) {
|
||||||
|
heartbeatLog.warn(
|
||||||
|
`Failed to emit stale agent assignment audit for ${options.agent.id}/${options.taskId}: ${error instanceof Error ? error.message : String(error)}`,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Find agents in `state="running"` that are not actually running and flip
|
* Find agents in `state="running"` that are not actually running and flip
|
||||||
* them to `"active"`. An agent is considered orphaned when either:
|
* them to `"active"`. An agent is considered orphaned when either:
|
||||||
@@ -1273,8 +1313,8 @@ export class HeartbeatMonitor {
|
|||||||
* logged but do not block the caller.
|
* logged but do not block the caller.
|
||||||
*
|
*
|
||||||
* Complements SelfHealingManager.recoverAgentsRunningOnInactiveTasks():
|
* Complements SelfHealingManager.recoverAgentsRunningOnInactiveTasks():
|
||||||
* heartbeat reconciliation handles stale/no-run conditions, while self-healing
|
* heartbeat reconciliation handles stale/no-run conditions and the prompt-critical
|
||||||
* handles task-column mismatches (for example running agents linked to todo tasks).
|
* parked todo/triage assignment drift before Reports Health Check renders.
|
||||||
*/
|
*/
|
||||||
private async reconcileOrphanedRunningAgents(): Promise<void> {
|
private async reconcileOrphanedRunningAgents(): Promise<void> {
|
||||||
try {
|
try {
|
||||||
@@ -1282,10 +1322,33 @@ export class HeartbeatMonitor {
|
|||||||
const now = Date.now();
|
const now = Date.now();
|
||||||
for (const agent of runningAgents) {
|
for (const agent of runningAgents) {
|
||||||
let reason: string | null = null;
|
let reason: string | null = null;
|
||||||
|
let clearTaskLink = false;
|
||||||
|
let taskIdToClear: string | null = null;
|
||||||
|
let parkedProof: AgentTaskLinkExecutionProof | null = null;
|
||||||
|
let linkedTask: TaskDetail | null = null;
|
||||||
const activeRun = await this.store.getActiveHeartbeatRun(agent.id);
|
const activeRun = await this.store.getActiveHeartbeatRun(agent.id);
|
||||||
if (!activeRun) {
|
if (!isEphemeralAgent(agent) && agent.taskId && this.taskStore) {
|
||||||
|
linkedTask = await this.taskStore.getTask(agent.taskId);
|
||||||
|
parkedProof = evaluateParkedAgentTaskLink({
|
||||||
|
agent,
|
||||||
|
linkedTask,
|
||||||
|
activeRun,
|
||||||
|
hasActiveAgentExecution: (agentId) => this.trackedAgents.has(agentId),
|
||||||
|
now,
|
||||||
|
});
|
||||||
|
/*
|
||||||
|
FNXC:AgentTaskStateDrift 2026-06-23-09:02:
|
||||||
|
Reports Health Check must not render a durable direct report as running a parked todo/triage task unless a fresh heartbeat run or tracked executor signal proves live execution. Clearing Agent.taskId here preserves overlapBlockedBy on the task row; the file-scope lease remains the scheduler's source of truth.
|
||||||
|
*/
|
||||||
|
if (isParkedTaskColumn(linkedTask) && !parkedProof.shouldPreserveParkedLink) {
|
||||||
|
reason = `parked ${linkedTask.column} task ${agent.taskId} without live execution proof`;
|
||||||
|
clearTaskLink = true;
|
||||||
|
taskIdToClear = agent.taskId;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (!reason && !activeRun) {
|
||||||
reason = "no active run";
|
reason = "no active run";
|
||||||
} else if (!this.trackedAgents.has(agent.id)) {
|
} else if (!reason && activeRun && !this.trackedAgents.has(agent.id)) {
|
||||||
const timeoutMs = this.resolveAgentConfig(agent.id).heartbeatTimeoutMs;
|
const timeoutMs = this.resolveAgentConfig(agent.id).heartbeatTimeoutMs;
|
||||||
const heartbeatAgeMs = getHeartbeatAgeMs(agent, now);
|
const heartbeatAgeMs = getHeartbeatAgeMs(agent, now);
|
||||||
// NOTE(FN-4278): this stale gate intentionally uses a per-run work-budget
|
// NOTE(FN-4278): this stale gate intentionally uses a per-run work-budget
|
||||||
@@ -1311,9 +1374,21 @@ export class HeartbeatMonitor {
|
|||||||
}
|
}
|
||||||
if (!reason) continue;
|
if (!reason) continue;
|
||||||
try {
|
try {
|
||||||
|
const staleAgentState = agent.state;
|
||||||
await this.store.updateAgentState(agent.id, "active");
|
await this.store.updateAgentState(agent.id, "active");
|
||||||
|
if (clearTaskLink) {
|
||||||
|
await this.store.syncExecutionTaskLink(agent.id, undefined);
|
||||||
|
await this.emitStaleAgentAssignmentAudit({
|
||||||
|
agent: { id: agent.id, state: staleAgentState },
|
||||||
|
taskId: taskIdToClear!,
|
||||||
|
linkedTask,
|
||||||
|
hadFreshRun: parkedProof?.hasFreshRun ?? false,
|
||||||
|
hadActiveExecution: parkedProof?.hasActiveExecution ?? false,
|
||||||
|
reason,
|
||||||
|
});
|
||||||
|
}
|
||||||
this.clearRunState(agent.id);
|
this.clearRunState(agent.id);
|
||||||
heartbeatLog.log(`Reconciled orphaned running agent ${agent.id} → active (${reason})`);
|
heartbeatLog.log(`Reconciled orphaned running agent ${agent.id} → active (${reason})${clearTaskLink ? "; stale task link cleared" : ""}`);
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
heartbeatLog.warn(`Failed to reconcile orphaned running agent ${agent.id}: ${err instanceof Error ? err.message : String(err)}`);
|
heartbeatLog.warn(`Failed to reconcile orphaned running agent ${agent.id}: ${err instanceof Error ? err.message : String(err)}`);
|
||||||
}
|
}
|
||||||
@@ -3234,8 +3309,36 @@ export class HeartbeatMonitor {
|
|||||||
const lastHeartbeatTs = report.lastHeartbeatAt ? Date.parse(report.lastHeartbeatAt) : NaN;
|
const lastHeartbeatTs = report.lastHeartbeatAt ? Date.parse(report.lastHeartbeatAt) : NaN;
|
||||||
const heartbeatAgeMs = Number.isFinite(lastHeartbeatTs) ? Math.max(0, now - lastHeartbeatTs) : Infinity;
|
const heartbeatAgeMs = Number.isFinite(lastHeartbeatTs) ? Math.max(0, now - lastHeartbeatTs) : Infinity;
|
||||||
|
|
||||||
|
let renderedState = report.state;
|
||||||
|
let renderedTask = report.taskId ?? "—";
|
||||||
|
let staleParkedAssignment = false;
|
||||||
|
if (report.state === "running" && !isEphemeralAgent(report) && report.taskId && this.taskStore) {
|
||||||
|
try {
|
||||||
|
const linkedTask = await this.taskStore.getTask(report.taskId);
|
||||||
|
if (isParkedTaskColumn(linkedTask)) {
|
||||||
|
const activeRun = await agentStore.getActiveHeartbeatRun(report.id);
|
||||||
|
const proof = evaluateParkedAgentTaskLink({
|
||||||
|
agent: report,
|
||||||
|
linkedTask,
|
||||||
|
activeRun,
|
||||||
|
hasActiveAgentExecution: (candidateId) => this.trackedAgents.has(candidateId),
|
||||||
|
now,
|
||||||
|
});
|
||||||
|
if (!proof.shouldPreserveParkedLink) {
|
||||||
|
staleParkedAssignment = true;
|
||||||
|
renderedState = "active";
|
||||||
|
renderedTask = `${report.taskId} (queued/no live run)`;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
} catch (error) {
|
||||||
|
heartbeatLog.warn(`[reports-health] failed to validate task link for ${report.id}/${report.taskId}: ${error instanceof Error ? error.message : String(error)}`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
let health = "healthy";
|
let health = "healthy";
|
||||||
if (report.state === "paused") {
|
if (staleParkedAssignment) {
|
||||||
|
health = "**stale** assignment";
|
||||||
|
} else if (report.state === "paused") {
|
||||||
health = report.pauseReason ? `paused (${report.pauseReason})` : "paused";
|
health = report.pauseReason ? `paused (${report.pauseReason})` : "paused";
|
||||||
} else if (report.state === "error") {
|
} else if (report.state === "error") {
|
||||||
health = "**stuck**";
|
health = "**stuck**";
|
||||||
@@ -3246,8 +3349,8 @@ export class HeartbeatMonitor {
|
|||||||
heartbeatLog.log(`[reports-health] stale report ${report.id} intervalSource=${intervalSource} staleThresholdMs=${staleThresholdMs} heartbeatAgeMs=${heartbeatAgeMs}`);
|
heartbeatLog.log(`[reports-health] stale report ${report.id} intervalSource=${intervalSource} staleThresholdMs=${staleThresholdMs} heartbeatAgeMs=${heartbeatAgeMs}`);
|
||||||
}
|
}
|
||||||
|
|
||||||
const task = report.taskId ?? "—";
|
const task = renderedTask;
|
||||||
const state = report.state;
|
const state = renderedState;
|
||||||
const heartbeat = formatRelativeTime(report.lastHeartbeatAt);
|
const heartbeat = formatRelativeTime(report.lastHeartbeatAt);
|
||||||
return `| ${report.name} | ${state} | ${task} | ${heartbeat} | ${health} |`;
|
return `| ${report.name} | ${state} | ${task} | ${heartbeat} | ${health} |`;
|
||||||
}));
|
}));
|
||||||
|
|||||||
@@ -515,6 +515,11 @@ export type DatabaseMutationType =
|
|||||||
| "task:resume-limbo-escalated"
|
| "task:resume-limbo-escalated"
|
||||||
/** Metadata: { taskId, executionAgeMs, graceMs, staleBindingAgeFloorMs, checkedOutBy, agentPresent, lastActivityMs, hasRecentRunAudit, worktree, branch, worktreeExists, signalReason } */
|
/** Metadata: { taskId, executionAgeMs, graceMs, staleBindingAgeFloorMs, checkedOutBy, agentPresent, lastActivityMs, hasRecentRunAudit, worktree, branch, worktreeExists, signalReason } */
|
||||||
| "task:reclaim-phantom-executor-binding"
|
| "task:reclaim-phantom-executor-binding"
|
||||||
|
/**
|
||||||
|
* FNXC:AgentTaskStateDrift 2026-06-23-08:50:
|
||||||
|
* Self-healing must leave file-scope lease queues intact while recording when stale durable Agent.taskId/state drift is cleared. Metadata: { agentId, taskId, taskColumn, agentState, status, blockedBy, overlapBlockedBy, hadFreshRun, hadActiveExecution, reason }.
|
||||||
|
*/
|
||||||
|
| "task:reconcile-stale-agent-assignment"
|
||||||
/** Metadata: { taskId, branch, worktree, checkedOutBy, executionStartedAt, executionAgeMs, graceMs, liveWorktreeBoundBranch, reason } */
|
/** Metadata: { taskId, branch, worktree, checkedOutBy, executionStartedAt, executionAgeMs, graceMs, liveWorktreeBoundBranch, reason } */
|
||||||
| "task:reclaim-self-owned-branch-conflict-no-action"
|
| "task:reclaim-self-owned-branch-conflict-no-action"
|
||||||
| "task:orphan-detected-no-action"
|
| "task:orphan-detected-no-action"
|
||||||
|
|||||||
@@ -375,6 +375,7 @@ export class InProcessRuntime
|
|||||||
maxWorktrees: this.config.maxWorktrees,
|
maxWorktrees: this.config.maxWorktrees,
|
||||||
semaphore: this.globalSemaphore,
|
semaphore: this.globalSemaphore,
|
||||||
agentStore: this.agentStore,
|
agentStore: this.agentStore,
|
||||||
|
hasActiveAgentExecution: (agentId: string) => this.heartbeatMonitor?.getTrackedAgents().includes(agentId) ?? false,
|
||||||
missionStore,
|
missionStore,
|
||||||
missionAutopilot,
|
missionAutopilot,
|
||||||
missionExecutionLoop,
|
missionExecutionLoop,
|
||||||
|
|||||||
@@ -36,6 +36,7 @@ import { UnlinkedMissionsAdvisoryReporter } from "./unlinked-missions-advisory-r
|
|||||||
import { createRunAuditor, generateSyntheticRunId } from "./run-audit.js";
|
import { createRunAuditor, generateSyntheticRunId } from "./run-audit.js";
|
||||||
import { isWorkflowColumnsEnabled, DEFAULT_WORKFLOW_POOL_ID } from "@fusion/core";
|
import { isWorkflowColumnsEnabled, DEFAULT_WORKFLOW_POOL_ID } from "@fusion/core";
|
||||||
import { runHoldReleaseSweep, type SlotReservation } from "./hold-release.js";
|
import { runHoldReleaseSweep, type SlotReservation } from "./hold-release.js";
|
||||||
|
import { evaluateParkedAgentTaskLink } from "./task-agent-sync.js";
|
||||||
|
|
||||||
function shouldRunWorkflowColumnScheduler(_settings: Settings): boolean {
|
function shouldRunWorkflowColumnScheduler(_settings: Settings): boolean {
|
||||||
/*
|
/*
|
||||||
@@ -446,6 +447,8 @@ export interface SchedulerOptions {
|
|||||||
semaphore?: AgentSemaphore;
|
semaphore?: AgentSemaphore;
|
||||||
/** Optional AgentStore for durable-agent state rollback during overlap requeue. */
|
/** Optional AgentStore for durable-agent state rollback during overlap requeue. */
|
||||||
agentStore?: AgentStore;
|
agentStore?: AgentStore;
|
||||||
|
/** Optional live executor signal that preserves parked durable-agent links while work is truly active. */
|
||||||
|
hasActiveAgentExecution?: (agentId: string) => boolean;
|
||||||
/** Called when scheduler starts a task */
|
/** Called when scheduler starts a task */
|
||||||
onSchedule?: (task: Task) => void;
|
onSchedule?: (task: Task) => void;
|
||||||
/** Called when a task is blocked by deps */
|
/** Called when a task is blocked by deps */
|
||||||
@@ -1068,13 +1071,29 @@ export class Scheduler {
|
|||||||
const agentStore = this.options.agentStore;
|
const agentStore = this.options.agentStore;
|
||||||
if (!agentStore) return;
|
if (!agentStore) return;
|
||||||
|
|
||||||
const runningAgents = await agentStore.listAgents({ state: "running", includeEphemeral: true });
|
const runningAgents = await agentStore.listAgents({ state: "running", includeEphemeral: false });
|
||||||
const linkedAgents = runningAgents.filter((agent) => agent.taskId === taskId);
|
const linkedAgents = runningAgents.filter((agent) => agent.taskId === taskId);
|
||||||
|
|
||||||
for (const agent of linkedAgents) {
|
for (const agent of linkedAgents) {
|
||||||
|
const activeRun = await agentStore.getActiveHeartbeatRun?.(agent.id);
|
||||||
|
const proof = evaluateParkedAgentTaskLink({
|
||||||
|
agent,
|
||||||
|
linkedTask: { column: "todo" } as Pick<Task, "column">,
|
||||||
|
activeRun,
|
||||||
|
hasActiveAgentExecution: this.options.hasActiveAgentExecution,
|
||||||
|
});
|
||||||
|
if (proof.shouldPreserveParkedLink) {
|
||||||
|
schedulerLog.log(
|
||||||
|
`Preserved running agent ${agent.id} for queued ${taskId}; live proof freshRun=${proof.hasFreshRun} activeExecution=${proof.hasActiveExecution}`,
|
||||||
|
);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
await agentStore.updateAgentState(agent.id, "active");
|
await agentStore.updateAgentState(agent.id, "active");
|
||||||
await agentStore.syncExecutionTaskLink(agent.id, undefined);
|
await agentStore.syncExecutionTaskLink(agent.id, undefined);
|
||||||
schedulerLog.log(`Rolled back running agent ${agent.id} after overlap requeue of ${taskId}`);
|
schedulerLog.log(
|
||||||
|
`Cleared stale running agent ${agent.id} after overlap requeue of ${taskId}; file-scope lease remains queued`,
|
||||||
|
);
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -2513,6 +2532,7 @@ export class Scheduler {
|
|||||||
blockedBy: null,
|
blockedBy: null,
|
||||||
overlapBlockedBy: overlappingTaskId,
|
overlapBlockedBy: overlappingTaskId,
|
||||||
});
|
});
|
||||||
|
await this.rollbackRunningAgentsForQueuedTodoTask(task.id);
|
||||||
await this.logDispatchQueuedReason(
|
await this.logDispatchQueuedReason(
|
||||||
task.id,
|
task.id,
|
||||||
`queued — blocked by active file-scope lease ${overlappingTaskId} (column=${activeLeaseColumn})`,
|
`queued — blocked by active file-scope lease ${overlappingTaskId} (column=${activeLeaseColumn})`,
|
||||||
|
|||||||
@@ -30,7 +30,7 @@ import { setImmediate as setImmediateCb } from "node:timers";
|
|||||||
import { existsSync, mkdirSync, readdirSync, readFileSync, realpathSync, rmSync, statSync, writeFileSync } from "node:fs";
|
import { existsSync, mkdirSync, readdirSync, readFileSync, realpathSync, rmSync, statSync, writeFileSync } from "node:fs";
|
||||||
import { tmpdir } from "node:os";
|
import { tmpdir } from "node:os";
|
||||||
import { isAbsolute, join, relative, resolve } from "node:path";
|
import { isAbsolute, join, relative, resolve } from "node:path";
|
||||||
import { IN_REVIEW_STALL_DEADLOCK_LOG_PREFIX, IN_REVIEW_STALL_LOG_PREFIX, IN_REVIEW_STALL_TERMINAL_LOG_PREFIX, allowsAutoMergeProcessing, countRecentIdenticalStallEntries, detectDependencyCycle, detectSelfDefeatingDependency, evaluateNoCommitsNoOpFinalize, getInReviewStalledSignal, getInReviewStallReason, getPrimaryPrInfo, getStalePausedReviewSignal, getStalePausedTodoSignal, getTaskHardMergeBlocker, getTaskMergeBlocker, isEphemeralAgent, isMergeRequestContractShadowEnabled, isWorkflowColumnsEnabled, isSharedBranchGroupMemberIntegration, parseExplicitDuplicateMarker, resolveMaxAutoMergeRetries, type AgentStore, type ChatStore, type MessageStore, type TaskStore, type Settings, type Task, type MergeDetails, type TaskPriority, type MergeResult } from "@fusion/core";
|
import { IN_REVIEW_STALL_DEADLOCK_LOG_PREFIX, IN_REVIEW_STALL_LOG_PREFIX, IN_REVIEW_STALL_TERMINAL_LOG_PREFIX, allowsAutoMergeProcessing, countRecentIdenticalStallEntries, detectDependencyCycle, detectSelfDefeatingDependency, evaluateNoCommitsNoOpFinalize, getInReviewStalledSignal, getInReviewStallReason, getPrimaryPrInfo, getStalePausedReviewSignal, getStalePausedTodoSignal, getTaskHardMergeBlocker, getTaskMergeBlocker, isEphemeralAgent, isMergeRequestContractShadowEnabled, isWorkflowColumnsEnabled, isSharedBranchGroupMemberIntegration, parseExplicitDuplicateMarker, resolveMaxAutoMergeRetries, type Agent, type AgentStore, type ChatStore, type MessageStore, type TaskStore, type Settings, type Task, type MergeDetails, type TaskPriority, type MergeResult } from "@fusion/core";
|
||||||
import type { MeshLeaseManager } from "./mesh-lease-manager.js";
|
import type { MeshLeaseManager } from "./mesh-lease-manager.js";
|
||||||
import { createLogger, schedulerLog } from "./logger.js";
|
import { createLogger, schedulerLog } from "./logger.js";
|
||||||
import { mergeEffectiveSettings } from "./effective-settings.js";
|
import { mergeEffectiveSettings } from "./effective-settings.js";
|
||||||
@@ -67,6 +67,7 @@ import {
|
|||||||
import type { GhostBugDecision } from "./triage-preflight.js";
|
import type { GhostBugDecision } from "./triage-preflight.js";
|
||||||
import { DependencyBlockedTodoReporter } from "./dependency-blocked-todo-reporter.js";
|
import { DependencyBlockedTodoReporter } from "./dependency-blocked-todo-reporter.js";
|
||||||
import { filterPathsByIgnoreList, getUnmetSchedulingDependencies, isCoordinationOnlyTask, pathsOverlap } from "./scheduler.js";
|
import { filterPathsByIgnoreList, getUnmetSchedulingDependencies, isCoordinationOnlyTask, pathsOverlap } from "./scheduler.js";
|
||||||
|
import { evaluateParkedAgentTaskLink, PARKED_AGENT_LINK_FRESH_RUN_MS } from "./task-agent-sync.js";
|
||||||
|
|
||||||
const log = createLogger("self-healing");
|
const log = createLogger("self-healing");
|
||||||
const worktreeMetadataReconcileLog = createLogger("worktree-metadata-reconcile");
|
const worktreeMetadataReconcileLog = createLogger("worktree-metadata-reconcile");
|
||||||
@@ -451,7 +452,7 @@ const DEFAULT_UNBACKED_MERGING_FANOUT_GRACE_MS = 60_000;
|
|||||||
const DURABLE_ERROR_RECOVERY_MAX_RETRIES = 5;
|
const DURABLE_ERROR_RECOVERY_MAX_RETRIES = 5;
|
||||||
const DURABLE_ERROR_RECOVERY_BASE_COOLDOWN_MS = 30_000;
|
const DURABLE_ERROR_RECOVERY_BASE_COOLDOWN_MS = 30_000;
|
||||||
const DURABLE_ERROR_RECOVERY_MAX_COOLDOWN_MS = 15 * 60_000;
|
const DURABLE_ERROR_RECOVERY_MAX_COOLDOWN_MS = 15 * 60_000;
|
||||||
const RUNNING_ON_INACTIVE_TASK_STALE_RUN_MS = 5 * 60_000;
|
const RUNNING_ON_INACTIVE_TASK_STALE_RUN_MS = PARKED_AGENT_LINK_FRESH_RUN_MS;
|
||||||
|
|
||||||
function bumpTaskPriority(priority: TaskPriority | undefined): TaskPriority {
|
function bumpTaskPriority(priority: TaskPriority | undefined): TaskPriority {
|
||||||
switch (priority ?? "normal") {
|
switch (priority ?? "normal") {
|
||||||
@@ -8744,6 +8745,44 @@ export class SelfHealingManager {
|
|||||||
return Math.min(exponential, DURABLE_ERROR_RECOVERY_MAX_COOLDOWN_MS);
|
return Math.min(exponential, DURABLE_ERROR_RECOVERY_MAX_COOLDOWN_MS);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private async emitStaleAgentAssignmentAudit(options: {
|
||||||
|
agent: Pick<Agent, "id" | "state">;
|
||||||
|
taskId: string;
|
||||||
|
linkedTask?: Task | null;
|
||||||
|
hadFreshRun: boolean;
|
||||||
|
hadActiveExecution: boolean;
|
||||||
|
reason: string;
|
||||||
|
}): Promise<void> {
|
||||||
|
try {
|
||||||
|
await createRunAuditor(this.store, {
|
||||||
|
runId: generateSyntheticRunId("self-healing-stale-agent-assignment", options.taskId),
|
||||||
|
agentId: "self-healing",
|
||||||
|
taskId: options.taskId,
|
||||||
|
taskLineageId: options.linkedTask?.lineageId,
|
||||||
|
phase: "reconcile-stale-agent-assignment",
|
||||||
|
}).database({
|
||||||
|
type: "task:reconcile-stale-agent-assignment" as DatabaseMutationType,
|
||||||
|
target: options.agent.id,
|
||||||
|
metadata: {
|
||||||
|
agentId: options.agent.id,
|
||||||
|
taskId: options.taskId,
|
||||||
|
taskColumn: options.linkedTask?.column ?? null,
|
||||||
|
agentState: options.agent.state,
|
||||||
|
status: options.linkedTask?.status ?? null,
|
||||||
|
blockedBy: options.linkedTask?.blockedBy ?? null,
|
||||||
|
overlapBlockedBy: options.linkedTask?.overlapBlockedBy ?? null,
|
||||||
|
hadFreshRun: options.hadFreshRun,
|
||||||
|
hadActiveExecution: options.hadActiveExecution,
|
||||||
|
reason: options.reason,
|
||||||
|
},
|
||||||
|
});
|
||||||
|
} catch (error) {
|
||||||
|
log.warn(
|
||||||
|
`Failed to emit stale agent assignment audit for ${options.agent.id}/${options.taskId}: ${error instanceof Error ? error.message : String(error)}`,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
async recoverAgentsRunningOnInactiveTasks(): Promise<number> {
|
async recoverAgentsRunningOnInactiveTasks(): Promise<number> {
|
||||||
const agentStore = this.options.agentStore;
|
const agentStore = this.options.agentStore;
|
||||||
if (!agentStore) {
|
if (!agentStore) {
|
||||||
@@ -8765,17 +8804,33 @@ export class SelfHealingManager {
|
|||||||
}
|
}
|
||||||
|
|
||||||
const activeRun = await agentStore.getActiveHeartbeatRun(agent.id);
|
const activeRun = await agentStore.getActiveHeartbeatRun(agent.id);
|
||||||
const runStartedAt = activeRun?.startedAt;
|
const proof = evaluateParkedAgentTaskLink({
|
||||||
const runAgeMs = runStartedAt ? now - Date.parse(runStartedAt) : Number.POSITIVE_INFINITY;
|
agent,
|
||||||
const hasFreshRun = Boolean(activeRun) && Number.isFinite(runAgeMs) && runAgeMs <= RUNNING_ON_INACTIVE_TASK_STALE_RUN_MS;
|
linkedTask: linkedTask ?? { column: "todo" } as Pick<Task, "column">,
|
||||||
if (hasFreshRun || this.options.hasActiveAgentExecution?.(agent.id) === true) {
|
activeRun,
|
||||||
|
hasActiveAgentExecution: this.options.hasActiveAgentExecution,
|
||||||
|
now,
|
||||||
|
});
|
||||||
|
if (proof.hasFreshRun || proof.hasActiveExecution) {
|
||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const reason = linkedTask
|
||||||
|
? `running durable agent linked to inactive ${linkedTask.column} task without live execution proof`
|
||||||
|
: "running durable agent linked to missing task without live execution proof";
|
||||||
|
const staleAgentState = agent.state;
|
||||||
await agentStore.updateAgentState(agent.id, "active");
|
await agentStore.updateAgentState(agent.id, "active");
|
||||||
await agentStore.syncExecutionTaskLink(agent.id, undefined);
|
await agentStore.syncExecutionTaskLink(agent.id, undefined);
|
||||||
|
await this.emitStaleAgentAssignmentAudit({
|
||||||
|
agent: { id: agent.id, state: staleAgentState },
|
||||||
|
taskId: agent.taskId,
|
||||||
|
linkedTask,
|
||||||
|
hadFreshRun: proof.hasFreshRun,
|
||||||
|
hadActiveExecution: proof.hasActiveExecution,
|
||||||
|
reason,
|
||||||
|
});
|
||||||
recoveredAgentIds.add(agent.id);
|
recoveredAgentIds.add(agent.id);
|
||||||
log.log(`Recovered running durable agent ${agent.id} on inactive task ${agent.taskId}`);
|
log.log(`Recovered running durable agent ${agent.id} on inactive task ${agent.taskId}; file-scope lease preserved when present`);
|
||||||
}
|
}
|
||||||
|
|
||||||
return recoveredAgentIds.size;
|
return recoveredAgentIds.size;
|
||||||
@@ -8800,6 +8855,8 @@ export class SelfHealingManager {
|
|||||||
const linkedTask = await this.store.getTask(linkedTaskId);
|
const linkedTask = await this.store.getTask(linkedTaskId);
|
||||||
let shouldClear = false;
|
let shouldClear = false;
|
||||||
let reason = "";
|
let reason = "";
|
||||||
|
let hadFreshRun = false;
|
||||||
|
let hadActiveExecution = false;
|
||||||
|
|
||||||
if (!linkedTask) {
|
if (!linkedTask) {
|
||||||
shouldClear = true;
|
shouldClear = true;
|
||||||
@@ -8812,13 +8869,18 @@ export class SelfHealingManager {
|
|||||||
reason = `linked task assigned to ${linkedTask.assignedAgentId}`;
|
reason = `linked task assigned to ${linkedTask.assignedAgentId}`;
|
||||||
} else if (linkedTask.column === "todo" || linkedTask.column === "triage") {
|
} else if (linkedTask.column === "todo" || linkedTask.column === "triage") {
|
||||||
const activeRun = await agentStore.getActiveHeartbeatRun(agent.id);
|
const activeRun = await agentStore.getActiveHeartbeatRun(agent.id);
|
||||||
const runStartedAt = activeRun?.startedAt;
|
const proof = evaluateParkedAgentTaskLink({
|
||||||
const runAgeMs = runStartedAt ? now - Date.parse(runStartedAt) : Number.POSITIVE_INFINITY;
|
agent,
|
||||||
const hasFreshRun = Boolean(activeRun) && Number.isFinite(runAgeMs) && runAgeMs <= RUNNING_ON_INACTIVE_TASK_STALE_RUN_MS;
|
linkedTask,
|
||||||
const hasActiveExecution = this.options.hasActiveAgentExecution?.(agent.id) === true;
|
activeRun,
|
||||||
if (!hasFreshRun && !hasActiveExecution) {
|
hasActiveAgentExecution: this.options.hasActiveAgentExecution,
|
||||||
|
now,
|
||||||
|
});
|
||||||
|
hadFreshRun = proof.hasFreshRun;
|
||||||
|
hadActiveExecution = proof.hasActiveExecution;
|
||||||
|
if (!proof.shouldPreserveParkedLink) {
|
||||||
shouldClear = true;
|
shouldClear = true;
|
||||||
reason = `linked task in queued column ${linkedTask.column} without fresh run`;
|
reason = `linked task in queued column ${linkedTask.column} without fresh run or active execution`;
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -8826,9 +8888,21 @@ export class SelfHealingManager {
|
|||||||
continue;
|
continue;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const staleAgentState = agent.state;
|
||||||
|
if (agent.state === "running") {
|
||||||
|
await agentStore.updateAgentState(agent.id, "active");
|
||||||
|
}
|
||||||
await agentStore.syncExecutionTaskLink(agent.id, undefined);
|
await agentStore.syncExecutionTaskLink(agent.id, undefined);
|
||||||
|
await this.emitStaleAgentAssignmentAudit({
|
||||||
|
agent: { id: agent.id, state: staleAgentState },
|
||||||
|
taskId: linkedTaskId,
|
||||||
|
linkedTask,
|
||||||
|
hadFreshRun,
|
||||||
|
hadActiveExecution,
|
||||||
|
reason,
|
||||||
|
});
|
||||||
clearedAgentIds.add(agent.id);
|
clearedAgentIds.add(agent.id);
|
||||||
log.log(`Cleared drifted durable agent task link for ${agent.id} (${linkedTaskId}): ${reason}`);
|
log.log(`Cleared drifted durable agent task link for ${agent.id} (${linkedTaskId}): ${reason}; file-scope lease preserved when present`);
|
||||||
}
|
}
|
||||||
|
|
||||||
log.log(`Recovered ${clearedAgentIds.size} drifted durable agent task link(s)`);
|
log.log(`Recovered ${clearedAgentIds.size} drifted durable agent task link(s)`);
|
||||||
|
|||||||
@@ -1,4 +1,51 @@
|
|||||||
import type { AgentStore, TaskStore } from "@fusion/core";
|
import type { Agent, AgentHeartbeatRun, AgentStore, Task, TaskStore } from "@fusion/core";
|
||||||
|
|
||||||
|
export const PARKED_AGENT_LINK_FRESH_RUN_MS = 5 * 60_000;
|
||||||
|
|
||||||
|
export interface AgentTaskLinkExecutionProof {
|
||||||
|
hasFreshRun: boolean;
|
||||||
|
hasActiveExecution: boolean;
|
||||||
|
shouldPreserveParkedLink: boolean;
|
||||||
|
runAgeMs: number;
|
||||||
|
}
|
||||||
|
|
||||||
|
export function hasFreshActiveHeartbeatRun(
|
||||||
|
activeRun: AgentHeartbeatRun | null | undefined,
|
||||||
|
now = Date.now(),
|
||||||
|
freshRunMs = PARKED_AGENT_LINK_FRESH_RUN_MS,
|
||||||
|
): { hasFreshRun: boolean; runAgeMs: number } {
|
||||||
|
const runStartedAt = activeRun?.startedAt;
|
||||||
|
const runAgeMs = runStartedAt ? now - Date.parse(runStartedAt) : Number.POSITIVE_INFINITY;
|
||||||
|
return {
|
||||||
|
hasFreshRun: Boolean(activeRun) && Number.isFinite(runAgeMs) && runAgeMs <= freshRunMs,
|
||||||
|
runAgeMs,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
|
export function isParkedTaskColumn(task: Pick<Task, "column"> | null | undefined): boolean {
|
||||||
|
return task?.column === "todo" || task?.column === "triage";
|
||||||
|
}
|
||||||
|
|
||||||
|
export function evaluateParkedAgentTaskLink(options: {
|
||||||
|
agent: Pick<Agent, "id" | "taskId">;
|
||||||
|
linkedTask: Pick<Task, "column"> | null | undefined;
|
||||||
|
activeRun?: AgentHeartbeatRun | null;
|
||||||
|
hasActiveAgentExecution?: (agentId: string) => boolean;
|
||||||
|
now?: number;
|
||||||
|
}): AgentTaskLinkExecutionProof {
|
||||||
|
const { hasFreshRun, runAgeMs } = hasFreshActiveHeartbeatRun(options.activeRun, options.now);
|
||||||
|
const hasActiveExecution = options.hasActiveAgentExecution?.(options.agent.id) === true;
|
||||||
|
/*
|
||||||
|
FNXC:AgentTaskStateDrift 2026-06-23-08:33:
|
||||||
|
Agent.taskId is a running assignment for parked todo/triage tasks only when the agent has live execution proof: a fresh active heartbeat run or an executor-active signal. File-scope overlapBlockedBy keeps the task queued but never proves the blocked task itself is executing.
|
||||||
|
*/
|
||||||
|
return {
|
||||||
|
hasFreshRun,
|
||||||
|
hasActiveExecution,
|
||||||
|
shouldPreserveParkedLink: isParkedTaskColumn(options.linkedTask) && (hasFreshRun || hasActiveExecution),
|
||||||
|
runAgeMs,
|
||||||
|
};
|
||||||
|
}
|
||||||
|
|
||||||
type LoggerLike = { log: (msg: string) => void; warn: (msg: string) => void };
|
type LoggerLike = { log: (msg: string) => void; warn: (msg: string) => void };
|
||||||
|
|
||||||
@@ -24,10 +71,22 @@ export function attachAgentLinkSync(opts: AttachAgentLinkSyncOptions): () => voi
|
|||||||
const linkedAgents = agents.filter((agent) => agent.taskId === task.id);
|
const linkedAgents = agents.filter((agent) => agent.taskId === task.id);
|
||||||
|
|
||||||
for (const agent of linkedAgents) {
|
for (const agent of linkedAgents) {
|
||||||
if ((to === "todo" || to === "triage") && opts.hasActiveAgentExecution?.(agent.id) === true) {
|
if (to === "todo" || to === "triage") {
|
||||||
continue;
|
const activeRun = await opts.agentStore.getActiveHeartbeatRun?.(agent.id);
|
||||||
|
const proof = evaluateParkedAgentTaskLink({
|
||||||
|
agent,
|
||||||
|
linkedTask: { column: to } as Pick<Task, "column">,
|
||||||
|
activeRun,
|
||||||
|
hasActiveAgentExecution: opts.hasActiveAgentExecution,
|
||||||
|
});
|
||||||
|
if (proof.shouldPreserveParkedLink) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
if (agent.state === "running") {
|
||||||
|
await opts.agentStore.updateAgentState(agent.id, "active");
|
||||||
|
}
|
||||||
await opts.agentStore.syncExecutionTaskLink(agent.id, undefined);
|
await opts.agentStore.syncExecutionTaskLink(agent.id, undefined);
|
||||||
logger.log(`taskAgentLinkSync: cleared agent ${agent.id} taskId from ${task.id} after move ${from} → ${to}`);
|
logger.log(`taskAgentLinkSync: cleared agent ${agent.id} taskId from ${task.id} after move ${from} → ${to}`);
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user