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

feat(FN-4132): sync task ownership on agent reassignment

Browse Source 
Adds a task reassignment sync helper to the core store, wires it into the system, and ships regression tests; includes a patch changeset for `@runfusion/fusion`.

Fusion-Task-Id: FN-4132

Fusion-Task-Lineage: 80ec3c62-7c09-49dd-9e49-74b203e48bf4
This commit is contained in:
Fusion
2026-05-12 10:04:56 -07:00
committed by gsxdsm
parent a289c00631
commit 6c1cb40c35
6 changed files with 264 additions and 11 deletions
Download Patch File Download Diff File Expand all files Collapse all files

129
packages/core/src/__tests__/store-persistence.test.ts
View File

@@ -1,4 +1,6 @@
import { describe, it, expect, beforeEach, afterEach } from "vitest";
import { AgentStore } from "../agent-store.js";
import { TaskStore } from "../store.js";
import { createTaskStoreTestHarness } from "./store-test-helpers.js";
describe("TaskStore", () => {
@@ -57,6 +59,133 @@ describe("TaskStore", () => {
});
});
describe("agent taskId sync on reassignment", () => {
it("reassignment clears the old agent taskId and sets the new agent taskId", async () => {
harness.store().close();
const store = new TaskStore(harness.rootDir(), harness.globalDir());
await store.init();
const agentStore = new AgentStore({ rootDir: store.getFusionDir() });
await agentStore.init();
try {
const task = await store.createTask({ description: "Reassignment target" });
const agentA = await agentStore.createAgent({ name: "Agent A", role: "executor" });
const agentB = await agentStore.createAgent({ name: "Agent B", role: "executor" });
await store.updateTask(task.id, { assignedAgentId: agentA.id });
expect((await agentStore.getAgent(agentA.id))?.taskId).toBe(task.id);
await store.updateTask(task.id, { assignedAgentId: agentB.id });
expect((await agentStore.getAgent(agentA.id))?.taskId).toBeUndefined();
expect((await agentStore.getAgent(agentB.id))?.taskId).toBe(task.id);
} finally {
agentStore.close();
store.close();
}
});
it("reassignment clears stale checkedOutBy when the outgoing agent held the lease", async () => {
harness.store().close();
const store = new TaskStore(harness.rootDir(), harness.globalDir());
await store.init();
const agentStore = new AgentStore({ rootDir: store.getFusionDir(), taskStore: store });
await agentStore.init();
try {
const task = await store.createTask({ description: "Checkout cleanup target" });
const agentA = await agentStore.createAgent({ name: "Agent A Checkout", role: "executor" });
const agentB = await agentStore.createAgent({ name: "Agent B Checkout", role: "executor" });
await store.updateTask(task.id, { assignedAgentId: agentA.id });
await agentStore.checkoutTask(agentA.id, task.id);
expect((await store.getTask(task.id)).checkedOutBy).toBe(agentA.id);
const updated = await store.updateTask(task.id, { assignedAgentId: agentB.id });
expect(updated.checkedOutBy).toBeUndefined();
} finally {
agentStore.close();
store.close();
}
});
it("unassignment clears the agent taskId", async () => {
harness.store().close();
const store = new TaskStore(harness.rootDir(), harness.globalDir());
await store.init();
const agentStore = new AgentStore({ rootDir: store.getFusionDir() });
await agentStore.init();
try {
const task = await store.createTask({ description: "Unassign target" });
const agent = await agentStore.createAgent({ name: "Sole Agent", role: "executor" });
await store.updateTask(task.id, { assignedAgentId: agent.id });
expect((await agentStore.getAgent(agent.id))?.taskId).toBe(task.id);
await store.updateTask(task.id, { assignedAgentId: null });
expect((await agentStore.getAgent(agent.id))?.taskId).toBeUndefined();
} finally {
agentStore.close();
store.close();
}
});
it("re-setting the same agent id is a no-op for agent task links", async () => {
harness.store().close();
const store = new TaskStore(harness.rootDir(), harness.globalDir());
await store.init();
const agentStore = new AgentStore({ rootDir: store.getFusionDir() });
await agentStore.init();
try {
const task = await store.createTask({ description: "Idempotent target" });
const agentA = await agentStore.createAgent({ name: "Agent Same", role: "executor" });
await store.updateTask(task.id, { assignedAgentId: agentA.id });
expect((await agentStore.getAgent(agentA.id))?.taskId).toBe(task.id);
await store.updateTask(task.id, { assignedAgentId: agentA.id });
expect((await agentStore.getAgent(agentA.id))?.taskId).toBe(task.id);
} finally {
agentStore.close();
store.close();
}
});
it("does not clear the outgoing agent taskId when it already moved to another task", async () => {
harness.store().close();
const store = new TaskStore(harness.rootDir(), harness.globalDir());
await store.init();
const agentStore = new AgentStore({ rootDir: store.getFusionDir() });
await agentStore.init();
try {
const task1 = await store.createTask({ description: "Race guard task 1" });
const task2 = await store.createTask({ description: "Race guard task 2" });
const agentA = await agentStore.createAgent({ name: "Agent Race A", role: "executor" });
const agentB = await agentStore.createAgent({ name: "Agent Race B", role: "executor" });
await store.updateTask(task1.id, { assignedAgentId: agentA.id });
expect((await agentStore.getAgent(agentA.id))?.taskId).toBe(task1.id);
expect((await agentStore.getAgent(agentB.id))?.taskId).toBeUndefined();
await agentStore.syncExecutionTaskLink(agentA.id, task2.id);
expect((await agentStore.getAgent(agentA.id))?.taskId).toBe(task2.id);
await store.updateTask(task1.id, { assignedAgentId: agentB.id });
expect((await agentStore.getAgent(agentA.id))?.taskId).toBe(task2.id);
expect((await agentStore.getAgent(agentB.id))?.taskId).toBe(task1.id);
} finally {
agentStore.close();
store.close();
}
});
});
describe("pausedByAgentId persistence", () => {
it("creates and lists a task with pausedByAgentId", async () => {
const task = await harness.store().createTask({ description: "Agent paused task" });

1
packages/core/src/__tests__/store-test-helpers.ts
View File

@@ -45,6 +45,7 @@ export function createTaskStoreTestHarness() {
globalDir: () => globalDir,
store: () => store,
beforeEach: async () => {
vi.useRealTimers();
rootDir = makeTmpDir();
globalDir = makeTmpDir();
store = new TaskStore(rootDir, globalDir, { inMemoryDb: true });

59
packages/core/src/store.ts
View File

@@ -3677,6 +3677,20 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
...(runContext ? { runContext } : {}),
});
}
if (assignmentChanged) {
this.syncAgentTaskLinkOnReassignment(id, previousAssignedAgentId, task.assignedAgentId);
if (task.checkedOutBy === previousAssignedAgentId) {
task.checkedOutBy = undefined;
task.checkedOutAt = undefined;
}
task.log.push({
timestamp: new Date().toISOString(),
action: `Agent task link synced: ${previousAssignedAgentId ?? "none"} → ${task.assignedAgentId ?? "none"}`,
...(runContext ? { runContext } : {}),
});
}
if (updates.pausedByAgentId === null) {
task.pausedByAgentId = undefined;
} else if (updates.pausedByAgentId !== undefined) {
@@ -4595,6 +4609,51 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
`).run(updatedAt, updatedAt, taskId);
}
/**
* Sync `agents.taskId` when {@link updateTask} reassigns a task.
*
* Uses direct SQL against the shared `agents` table instead of AgentStore to
* avoid a circular dependency while keeping the column and JSON data blob in
* lockstep. Clearing the previous agent is race-guarded with `WHERE id = ?
* AND taskId = ?` so we do not clobber an agent that already moved on to a
* different task.
*/
private syncAgentTaskLinkOnReassignment(
taskId: string,
previousAgentId: string | undefined,
newAgentId: string | undefined,
): void {
const updatedAt = new Date().toISOString();
if (previousAgentId) {
this.db.prepare(`
UPDATE agents
SET
taskId = NULL,
updatedAt = ?,
data = CASE
WHEN json_valid(data) THEN json_set(json_remove(data, '$.taskId'), '$.updatedAt', ?)
ELSE data
END
WHERE id = ? AND taskId = ?
`).run(updatedAt, updatedAt, previousAgentId, taskId);
}
if (newAgentId) {
this.db.prepare(`
UPDATE agents
SET
taskId = ?,
updatedAt = ?,
data = CASE
WHEN json_valid(data) THEN json_set(data, '$.taskId', ?, '$.updatedAt', ?)
ELSE data
END
WHERE id = ?
`).run(taskId, updatedAt, taskId, updatedAt, newAgentId);
}
}
/**
* Clean up the git branch associated with a task.
*