feat(FN-1255): add comment-aware heartbeat wakes and blocked-state dedup
- Add BlockedStateSnapshot typing/export and AgentStore persistence APIs for last blocked heartbeat state - Deduplicate blocked-task heartbeat comments using blockedBy + context hash, and clear blocked snapshots when tasks are no longer blocked - Thread triggeringCommentIds/triggeringCommentType through heartbeat execution, wake context, scheduler assignment triggers, and runtime wiring - Trigger immediate heartbeat runs from task/steering comment routes for assigned immediate-response agents, with validation for comment wake fields on /api/agents/:id/runs - Expand core, engine, and dashboard tests to cover blocked dedup logic, comment-triggered wakes, validation, and skip scenarios
This commit is contained in:
@@ -1753,6 +1753,50 @@ describe("AgentStore", () => {
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
// ── blocked state persistence ─────────────────────────────────────
|
||||||
|
|
||||||
|
describe("blocked state persistence", () => {
|
||||||
|
it("roundtrips last blocked state via set/get", async () => {
|
||||||
|
const agent = await store.createAgent({ name: "BlockedState", role: "executor" });
|
||||||
|
|
||||||
|
const snapshot = {
|
||||||
|
taskId: "FN-123",
|
||||||
|
blockedBy: "FN-122",
|
||||||
|
recordedAt: new Date().toISOString(),
|
||||||
|
contextHash: "abc123hash",
|
||||||
|
};
|
||||||
|
|
||||||
|
await store.setLastBlockedState(agent.id, snapshot);
|
||||||
|
const loaded = await store.getLastBlockedState(agent.id);
|
||||||
|
|
||||||
|
expect(loaded).toEqual(snapshot);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("returns null when no blocked-state file exists", async () => {
|
||||||
|
const agent = await store.createAgent({ name: "NoBlockedState", role: "executor" });
|
||||||
|
|
||||||
|
const loaded = await store.getLastBlockedState(agent.id);
|
||||||
|
expect(loaded).toBeNull();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("clearLastBlockedState removes persisted snapshot", async () => {
|
||||||
|
const agent = await store.createAgent({ name: "ClearBlockedState", role: "executor" });
|
||||||
|
|
||||||
|
await store.setLastBlockedState(agent.id, {
|
||||||
|
taskId: "FN-999",
|
||||||
|
blockedBy: "FN-998",
|
||||||
|
recordedAt: new Date().toISOString(),
|
||||||
|
contextHash: "will-clear",
|
||||||
|
});
|
||||||
|
|
||||||
|
await store.clearLastBlockedState(agent.id);
|
||||||
|
const loaded = await store.getLastBlockedState(agent.id);
|
||||||
|
|
||||||
|
expect(loaded).toBeNull();
|
||||||
|
expect(existsSync(join(rootDir, "agents", `${agent.id}-last-blocked.json`))).toBe(false);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
// ── getAgentDetail ────────────────────────────────────────────────
|
// ── getAgentDetail ────────────────────────────────────────────────
|
||||||
|
|
||||||
describe("getAgentDetail", () => {
|
describe("getAgentDetail", () => {
|
||||||
|
|||||||
@@ -28,6 +28,7 @@ import type {
|
|||||||
AgentApiKeyCreateResult,
|
AgentApiKeyCreateResult,
|
||||||
AgentHeartbeatEvent,
|
AgentHeartbeatEvent,
|
||||||
AgentHeartbeatRun,
|
AgentHeartbeatRun,
|
||||||
|
BlockedStateSnapshot,
|
||||||
AgentDetail,
|
AgentDetail,
|
||||||
AgentBudgetConfig,
|
AgentBudgetConfig,
|
||||||
AgentBudgetStatus,
|
AgentBudgetStatus,
|
||||||
@@ -1094,6 +1095,7 @@ export class AgentStore extends EventEmitter {
|
|||||||
const agentPath = join(this.agentsDir, `${agentId}.json`);
|
const agentPath = join(this.agentsDir, `${agentId}.json`);
|
||||||
const heartbeatPath = join(this.agentsDir, `${agentId}-heartbeats.jsonl`);
|
const heartbeatPath = join(this.agentsDir, `${agentId}-heartbeats.jsonl`);
|
||||||
const revisionsPath = this.getConfigRevisionsPath(agentId);
|
const revisionsPath = this.getConfigRevisionsPath(agentId);
|
||||||
|
const blockedStatePath = this.getLastBlockedStatePath(agentId);
|
||||||
|
|
||||||
// Verify agent exists
|
// Verify agent exists
|
||||||
const agent = await this.getAgent(agentId);
|
const agent = await this.getAgent(agentId);
|
||||||
@@ -1105,6 +1107,7 @@ export class AgentStore extends EventEmitter {
|
|||||||
await unlink(agentPath).catch(() => {});
|
await unlink(agentPath).catch(() => {});
|
||||||
await unlink(heartbeatPath).catch(() => {});
|
await unlink(heartbeatPath).catch(() => {});
|
||||||
await unlink(revisionsPath).catch(() => {});
|
await unlink(revisionsPath).catch(() => {});
|
||||||
|
await unlink(blockedStatePath).catch(() => {});
|
||||||
|
|
||||||
// Clean up sessions and runs directories
|
// Clean up sessions and runs directories
|
||||||
const { rm } = await import("node:fs/promises");
|
const { rm } = await import("node:fs/promises");
|
||||||
@@ -1552,6 +1555,41 @@ export class AgentStore extends EventEmitter {
|
|||||||
.slice(0, limit);
|
.slice(0, limit);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Get the most recently persisted blocked-task dedup state for an agent.
|
||||||
|
*/
|
||||||
|
async getLastBlockedState(agentId: string): Promise<BlockedStateSnapshot | null> {
|
||||||
|
const blockedStatePath = this.getLastBlockedStatePath(agentId);
|
||||||
|
try {
|
||||||
|
const content = await readFile(blockedStatePath, "utf-8");
|
||||||
|
return JSON.parse(content) as BlockedStateSnapshot;
|
||||||
|
} catch (err) {
|
||||||
|
if ((err as NodeJS.ErrnoException).code === "ENOENT") {
|
||||||
|
return null;
|
||||||
|
}
|
||||||
|
throw err;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Persist the latest blocked-task dedup state for an agent.
|
||||||
|
*/
|
||||||
|
async setLastBlockedState(agentId: string, state: BlockedStateSnapshot): Promise<void> {
|
||||||
|
await this.withLock(agentId, async () => {
|
||||||
|
const blockedStatePath = this.getLastBlockedStatePath(agentId);
|
||||||
|
await writeFile(blockedStatePath, JSON.stringify(state, null, 2));
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Clear any persisted blocked-task dedup state for an agent.
|
||||||
|
*/
|
||||||
|
async clearLastBlockedState(agentId: string): Promise<void> {
|
||||||
|
await this.withLock(agentId, async () => {
|
||||||
|
await unlink(this.getLastBlockedStatePath(agentId)).catch(() => {});
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
// ─────────────────────────────────────────────────────────────────────────
|
// ─────────────────────────────────────────────────────────────────────────
|
||||||
// Private helpers
|
// Private helpers
|
||||||
// ─────────────────────────────────────────────────────────────────────────
|
// ─────────────────────────────────────────────────────────────────────────
|
||||||
@@ -1783,6 +1821,10 @@ export class AgentStore extends EventEmitter {
|
|||||||
return join(this.agentsDir, `${agentId}-keys.jsonl`);
|
return join(this.agentsDir, `${agentId}-keys.jsonl`);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
private getLastBlockedStatePath(agentId: string): string {
|
||||||
|
return join(this.agentsDir, `${agentId}-last-blocked.json`);
|
||||||
|
}
|
||||||
|
|
||||||
private async readApiKeys(agentId: string): Promise<AgentApiKey[]> {
|
private async readApiKeys(agentId: string): Promise<AgentApiKey[]> {
|
||||||
const keyPath = this.getApiKeysPath(agentId);
|
const keyPath = this.getApiKeysPath(agentId);
|
||||||
if (!existsSync(keyPath)) {
|
if (!existsSync(keyPath)) {
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
export { COLUMNS, COLUMN_LABELS, COLUMN_DESCRIPTIONS, VALID_TRANSITIONS, DEFAULT_SETTINGS, DEFAULT_GLOBAL_SETTINGS, DEFAULT_PROJECT_SETTINGS, GLOBAL_SETTINGS_KEYS, PROJECT_SETTINGS_KEYS, THINKING_LEVELS, THEME_MODES, COLOR_THEMES, WORKFLOW_STEP_TEMPLATES, AGENT_PERMISSIONS, agentToConfigSnapshot, diffConfigSnapshots, CheckoutConflictError } from "./types.js";
|
export { COLUMNS, COLUMN_LABELS, COLUMN_DESCRIPTIONS, VALID_TRANSITIONS, DEFAULT_SETTINGS, DEFAULT_GLOBAL_SETTINGS, DEFAULT_PROJECT_SETTINGS, GLOBAL_SETTINGS_KEYS, PROJECT_SETTINGS_KEYS, THINKING_LEVELS, THEME_MODES, COLOR_THEMES, WORKFLOW_STEP_TEMPLATES, AGENT_PERMISSIONS, agentToConfigSnapshot, diffConfigSnapshots, CheckoutConflictError } from "./types.js";
|
||||||
export type { Column, IssueInfo, IssueState, PrInfo, PrStatus, Task, TaskAttachment, TaskComment, TaskCommentInput, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskCreateInput, TaskDetail, InboxTask, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, MergeDetails, MergeResult, Settings, GlobalSettings, ProjectSettings, SettingsScope, TaskStep, StepStatus, TaskLogEntry, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, OrgTreeNode, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentApiKey, AgentApiKeyCreateResult, AgentCapability, AgentPromptTemplate, AgentPromptsConfig, AgentPermission, TaskAssignSource, AgentAccessState, AgentHeartbeatConfig, AgentBudgetConfig, AgentBudgetStatus, InstructionsBundleConfig, MessageResponseMode, AgentHeartbeatEvent, AgentHeartbeatRun, HeartbeatInvocationSource, AgentTaskSession, AgentRating, AgentRatingSummary, AgentRatingInput, AgentConfigSnapshot, RevisionFieldDiff, AgentConfigRevision, AgentStats, ReflectionTrigger, ReflectionMetrics, AgentReflection, AgentPerformanceSummary, NtfyNotificationEvent, SteeringComment, ParticipantType, MessageType, Message, MessageCreateInput, MessageFilter, Mailbox, CheckoutLease } from "./types.js";
|
export type { Column, IssueInfo, IssueState, PrInfo, PrStatus, Task, TaskAttachment, TaskComment, TaskCommentInput, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskCreateInput, TaskDetail, InboxTask, AgentLogEntry, AgentLogType, AgentRole, BoardConfig, MergeDetails, MergeResult, Settings, GlobalSettings, ProjectSettings, SettingsScope, TaskStep, StepStatus, TaskLogEntry, ActivityLogEntry, ActivityEventType, ThinkingLevel, ThemeMode, ColorTheme, PlanningQuestion, PlanningSummary, PlanningResponse, PlanningQuestionType, ArchivedTaskEntry, BatchStatusRequest, BatchStatusResponse, BatchStatusEntry, BatchStatusResult, ModelPreset, WorkflowStep, WorkflowStepMode, WorkflowStepPhase, WorkflowStepInput, WorkflowStepResult, WorkflowStepTemplate, Agent, OrgTreeNode, AgentState, AgentDetail, AgentCreateInput, AgentUpdateInput, AgentApiKey, AgentApiKeyCreateResult, AgentCapability, AgentPromptTemplate, AgentPromptsConfig, AgentPermission, TaskAssignSource, AgentAccessState, AgentHeartbeatConfig, AgentBudgetConfig, AgentBudgetStatus, InstructionsBundleConfig, MessageResponseMode, AgentHeartbeatEvent, AgentHeartbeatRun, BlockedStateSnapshot, HeartbeatInvocationSource, AgentTaskSession, AgentRating, AgentRatingSummary, AgentRatingInput, AgentConfigSnapshot, RevisionFieldDiff, AgentConfigRevision, AgentStats, ReflectionTrigger, ReflectionMetrics, AgentReflection, AgentPerformanceSummary, NtfyNotificationEvent, SteeringComment, ParticipantType, MessageType, Message, MessageCreateInput, MessageFilter, Mailbox, CheckoutLease } from "./types.js";
|
||||||
export { AGENT_VALID_TRANSITIONS } from "./types.js";
|
export { AGENT_VALID_TRANSITIONS } from "./types.js";
|
||||||
export {
|
export {
|
||||||
BUILTIN_AGENT_PROMPTS,
|
BUILTIN_AGENT_PROMPTS,
|
||||||
|
|||||||
@@ -1719,6 +1719,18 @@ export interface AgentHeartbeatEvent {
|
|||||||
/** What triggered a heartbeat run */
|
/** What triggered a heartbeat run */
|
||||||
export type HeartbeatInvocationSource = "on_demand" | "timer" | "assignment" | "automation";
|
export type HeartbeatInvocationSource = "on_demand" | "timer" | "assignment" | "automation";
|
||||||
|
|
||||||
|
/** Snapshot of the last blocked state for a task, used for dedup comparison. */
|
||||||
|
export interface BlockedStateSnapshot {
|
||||||
|
/** The task ID that was blocked */
|
||||||
|
taskId: string;
|
||||||
|
/** What the task was blocked by (dependency IDs, overlapping task ID) */
|
||||||
|
blockedBy: string;
|
||||||
|
/** ISO-8601 timestamp when this blocked state was recorded */
|
||||||
|
recordedAt: string;
|
||||||
|
/** Hash of relevant context at the time (comment count, last comment ID) */
|
||||||
|
contextHash: string;
|
||||||
|
}
|
||||||
|
|
||||||
/** A continuous heartbeat session/run for an agent */
|
/** A continuous heartbeat session/run for an agent */
|
||||||
export interface AgentHeartbeatRun {
|
export interface AgentHeartbeatRun {
|
||||||
/** Unique identifier for this run */
|
/** Unique identifier for this run */
|
||||||
@@ -1747,7 +1759,10 @@ export interface AgentHeartbeatRun {
|
|||||||
usageJson?: { inputTokens: number; outputTokens: number; cachedTokens: number };
|
usageJson?: { inputTokens: number; outputTokens: number; cachedTokens: number };
|
||||||
/** Structured result from the run */
|
/** Structured result from the run */
|
||||||
resultJson?: Record<string, unknown>;
|
resultJson?: Record<string, unknown>;
|
||||||
/** Snapshot of context at run start (taskId, projectId, etc.) */
|
/** Snapshot of context at run start (taskId, projectId, etc.).
|
||||||
|
* May include optional comment-wake fields:
|
||||||
|
* - `triggeringCommentIds?: string[]`
|
||||||
|
* - `triggeringCommentType?: "steering" | "task" | "pr"` */
|
||||||
contextSnapshot?: Record<string, unknown>;
|
contextSnapshot?: Record<string, unknown>;
|
||||||
/** Excerpt of stdout output */
|
/** Excerpt of stdout output */
|
||||||
stdoutExcerpt?: string;
|
stdoutExcerpt?: string;
|
||||||
|
|||||||
@@ -3149,6 +3149,158 @@ describe("Pause/Unpause endpoints", () => {
|
|||||||
expect(store.addTaskComment).toHaveBeenCalledWith("KB-001", "Hello", "user");
|
expect(store.addTaskComment).toHaveBeenCalledWith("KB-001", "Hello", "user");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("POST /tasks/:id/comments — triggers immediate heartbeat wake for assigned agent", async () => {
|
||||||
|
const tempDir = mkdtempSync(join(tmpdir(), "kb-routes-comment-heartbeat-"));
|
||||||
|
const fusionDir = join(tempDir, ".fusion");
|
||||||
|
mkdirSync(fusionDir, { recursive: true });
|
||||||
|
|
||||||
|
try {
|
||||||
|
const { AgentStore } = await import("@fusion/core");
|
||||||
|
const agentStore = new AgentStore({ rootDir: fusionDir });
|
||||||
|
await agentStore.init();
|
||||||
|
const agent = await agentStore.createAgent({ name: "Wake Agent", role: "executor" });
|
||||||
|
await agentStore.updateAgent(agent.id, {
|
||||||
|
runtimeConfig: { messageResponseMode: "immediate" },
|
||||||
|
});
|
||||||
|
|
||||||
|
const heartbeatMonitor = {
|
||||||
|
executeHeartbeat: vi.fn().mockResolvedValue({ id: "run-1" }),
|
||||||
|
};
|
||||||
|
|
||||||
|
const updatedTask = {
|
||||||
|
...FAKE_TASK_DETAIL,
|
||||||
|
id: "KB-001",
|
||||||
|
assignedAgentId: agent.id,
|
||||||
|
comments: [{ id: "comment-1", text: "Hello", author: "user", createdAt: "2026-01-01T00:00:00.000Z" }],
|
||||||
|
};
|
||||||
|
|
||||||
|
const store = createMockStore({
|
||||||
|
addTaskComment: vi.fn().mockResolvedValue(updatedTask),
|
||||||
|
getFusionDir: vi.fn().mockReturnValue(fusionDir),
|
||||||
|
});
|
||||||
|
|
||||||
|
const app = express();
|
||||||
|
app.use(express.json());
|
||||||
|
app.use("/api", createApiRoutes(store, { heartbeatMonitor } as any));
|
||||||
|
|
||||||
|
const res = await REQUEST(app, "POST", "/api/tasks/KB-001/comments", JSON.stringify({ text: "Hello" }), {
|
||||||
|
"Content-Type": "application/json",
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(res.status).toBe(200);
|
||||||
|
await vi.waitFor(() => {
|
||||||
|
expect(heartbeatMonitor.executeHeartbeat).toHaveBeenCalledWith(expect.objectContaining({
|
||||||
|
agentId: agent.id,
|
||||||
|
source: "on_demand",
|
||||||
|
taskId: "KB-001",
|
||||||
|
triggeringCommentIds: ["comment-1"],
|
||||||
|
triggeringCommentType: "task",
|
||||||
|
}));
|
||||||
|
}, { timeout: 1000 });
|
||||||
|
} finally {
|
||||||
|
rmSync(tempDir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
it("POST /tasks/:id/comments — skips heartbeat wake when task has no assigned agent", async () => {
|
||||||
|
const heartbeatMonitor = {
|
||||||
|
executeHeartbeat: vi.fn().mockResolvedValue({ id: "run-1" }),
|
||||||
|
};
|
||||||
|
const updatedTask = {
|
||||||
|
...FAKE_TASK_DETAIL,
|
||||||
|
id: "KB-001",
|
||||||
|
assignedAgentId: undefined,
|
||||||
|
comments: [{ id: "comment-1", text: "Hello", author: "user", createdAt: "2026-01-01T00:00:00.000Z" }],
|
||||||
|
};
|
||||||
|
|
||||||
|
const store = createMockStore({
|
||||||
|
addTaskComment: vi.fn().mockResolvedValue(updatedTask),
|
||||||
|
});
|
||||||
|
|
||||||
|
const app = express();
|
||||||
|
app.use(express.json());
|
||||||
|
app.use("/api", createApiRoutes(store, { heartbeatMonitor } as any));
|
||||||
|
|
||||||
|
const res = await REQUEST(app, "POST", "/api/tasks/KB-001/comments", JSON.stringify({ text: "Hello" }), {
|
||||||
|
"Content-Type": "application/json",
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(res.status).toBe(200);
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||||
|
expect(heartbeatMonitor.executeHeartbeat).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("POST /tasks/:id/comments — succeeds without heartbeat monitor when task is assigned", async () => {
|
||||||
|
const updatedTask = {
|
||||||
|
...FAKE_TASK_DETAIL,
|
||||||
|
id: "KB-001",
|
||||||
|
assignedAgentId: "agent-123",
|
||||||
|
comments: [{ id: "comment-1", text: "Hello", author: "user", createdAt: "2026-01-01T00:00:00.000Z" }],
|
||||||
|
};
|
||||||
|
|
||||||
|
const store = createMockStore({
|
||||||
|
addTaskComment: vi.fn().mockResolvedValue(updatedTask),
|
||||||
|
});
|
||||||
|
|
||||||
|
const app = express();
|
||||||
|
app.use(express.json());
|
||||||
|
app.use("/api", createApiRoutes(store));
|
||||||
|
|
||||||
|
const res = await REQUEST(app, "POST", "/api/tasks/KB-001/comments", JSON.stringify({ text: "Hello" }), {
|
||||||
|
"Content-Type": "application/json",
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(res.status).toBe(200);
|
||||||
|
expect(store.addTaskComment).toHaveBeenCalledWith("KB-001", "Hello", "user");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("POST /tasks/:id/comments — skips heartbeat wake when an active run already exists", async () => {
|
||||||
|
const tempDir = mkdtempSync(join(tmpdir(), "kb-routes-comment-active-run-"));
|
||||||
|
const fusionDir = join(tempDir, ".fusion");
|
||||||
|
mkdirSync(fusionDir, { recursive: true });
|
||||||
|
|
||||||
|
try {
|
||||||
|
const { AgentStore } = await import("@fusion/core");
|
||||||
|
const agentStore = new AgentStore({ rootDir: fusionDir });
|
||||||
|
await agentStore.init();
|
||||||
|
const agent = await agentStore.createAgent({ name: "Active Run Agent", role: "executor" });
|
||||||
|
await agentStore.updateAgent(agent.id, {
|
||||||
|
runtimeConfig: { messageResponseMode: "immediate" },
|
||||||
|
});
|
||||||
|
await agentStore.startHeartbeatRun(agent.id);
|
||||||
|
|
||||||
|
const heartbeatMonitor = {
|
||||||
|
executeHeartbeat: vi.fn().mockResolvedValue({ id: "run-1" }),
|
||||||
|
};
|
||||||
|
|
||||||
|
const updatedTask = {
|
||||||
|
...FAKE_TASK_DETAIL,
|
||||||
|
id: "KB-001",
|
||||||
|
assignedAgentId: agent.id,
|
||||||
|
comments: [{ id: "comment-1", text: "Hello", author: "user", createdAt: "2026-01-01T00:00:00.000Z" }],
|
||||||
|
};
|
||||||
|
|
||||||
|
const store = createMockStore({
|
||||||
|
addTaskComment: vi.fn().mockResolvedValue(updatedTask),
|
||||||
|
getFusionDir: vi.fn().mockReturnValue(fusionDir),
|
||||||
|
});
|
||||||
|
|
||||||
|
const app = express();
|
||||||
|
app.use(express.json());
|
||||||
|
app.use("/api", createApiRoutes(store, { heartbeatMonitor } as any));
|
||||||
|
|
||||||
|
const res = await REQUEST(app, "POST", "/api/tasks/KB-001/comments", JSON.stringify({ text: "Hello" }), {
|
||||||
|
"Content-Type": "application/json",
|
||||||
|
});
|
||||||
|
|
||||||
|
expect(res.status).toBe(200);
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||||
|
expect(heartbeatMonitor.executeHeartbeat).not.toHaveBeenCalled();
|
||||||
|
} finally {
|
||||||
|
rmSync(tempDir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
it("PATCH /tasks/:id/comments/:commentId — updates a task comment", async () => {
|
it("PATCH /tasks/:id/comments/:commentId — updates a task comment", async () => {
|
||||||
const updatedTask = { ...FAKE_TASK_DETAIL, comments: [{ id: "c1", text: "Updated", author: "user", createdAt: "2026-01-01T00:00:00.000Z", updatedAt: "2026-01-01T00:01:00.000Z" }] };
|
const updatedTask = { ...FAKE_TASK_DETAIL, comments: [{ id: "c1", text: "Updated", author: "user", createdAt: "2026-01-01T00:00:00.000Z", updatedAt: "2026-01-01T00:01:00.000Z" }] };
|
||||||
const store = createMockStore({ updateTaskComment: vi.fn().mockResolvedValue(updatedTask) });
|
const store = createMockStore({ updateTaskComment: vi.fn().mockResolvedValue(updatedTask) });
|
||||||
@@ -3208,6 +3360,107 @@ describe("Pause/Unpause endpoints", () => {
|
|||||||
);
|
);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("triggers immediate heartbeat wake for assigned agent", async () => {
|
||||||
|
const tempDir = mkdtempSync(join(tmpdir(), "kb-routes-steer-heartbeat-"));
|
||||||
|
const fusionDir = join(tempDir, ".fusion");
|
||||||
|
mkdirSync(fusionDir, { recursive: true });
|
||||||
|
|
||||||
|
try {
|
||||||
|
const { AgentStore } = await import("@fusion/core");
|
||||||
|
const agentStore = new AgentStore({ rootDir: fusionDir });
|
||||||
|
await agentStore.init();
|
||||||
|
const agent = await agentStore.createAgent({ name: "Steer Wake Agent", role: "executor" });
|
||||||
|
await agentStore.updateAgent(agent.id, {
|
||||||
|
runtimeConfig: { messageResponseMode: "immediate" },
|
||||||
|
});
|
||||||
|
|
||||||
|
const heartbeatMonitor = {
|
||||||
|
executeHeartbeat: vi.fn().mockResolvedValue({ id: "run-1" }),
|
||||||
|
};
|
||||||
|
|
||||||
|
const steeredTask = {
|
||||||
|
...FAKE_TASK_DETAIL,
|
||||||
|
id: "KB-001",
|
||||||
|
assignedAgentId: agent.id,
|
||||||
|
steeringComments: [{ id: "steer-1", text: "Please handle edge case", author: "user", createdAt: "2026-01-01T00:00:00.000Z" }],
|
||||||
|
};
|
||||||
|
(store.addSteeringComment as ReturnType<typeof vi.fn>).mockResolvedValue(steeredTask);
|
||||||
|
(store.getFusionDir as any) = vi.fn().mockReturnValue(fusionDir);
|
||||||
|
|
||||||
|
const app = express();
|
||||||
|
app.use(express.json());
|
||||||
|
app.use("/api", createApiRoutes(store, { heartbeatMonitor } as any));
|
||||||
|
|
||||||
|
const res = await REQUEST(
|
||||||
|
app,
|
||||||
|
"POST",
|
||||||
|
"/api/tasks/KB-001/steer",
|
||||||
|
JSON.stringify({ text: "Please handle edge case" }),
|
||||||
|
{ "Content-Type": "application/json" },
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(res.status).toBe(200);
|
||||||
|
await vi.waitFor(() => {
|
||||||
|
expect(heartbeatMonitor.executeHeartbeat).toHaveBeenCalledWith(expect.objectContaining({
|
||||||
|
agentId: agent.id,
|
||||||
|
source: "on_demand",
|
||||||
|
taskId: "KB-001",
|
||||||
|
triggeringCommentIds: ["steer-1"],
|
||||||
|
triggeringCommentType: "steering",
|
||||||
|
}));
|
||||||
|
}, { timeout: 1000 });
|
||||||
|
} finally {
|
||||||
|
rmSync(tempDir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
|
it("skips heartbeat wake when assigned agent is not in immediate response mode", async () => {
|
||||||
|
const tempDir = mkdtempSync(join(tmpdir(), "kb-routes-steer-non-immediate-"));
|
||||||
|
const fusionDir = join(tempDir, ".fusion");
|
||||||
|
mkdirSync(fusionDir, { recursive: true });
|
||||||
|
|
||||||
|
try {
|
||||||
|
const { AgentStore } = await import("@fusion/core");
|
||||||
|
const agentStore = new AgentStore({ rootDir: fusionDir });
|
||||||
|
await agentStore.init();
|
||||||
|
const agent = await agentStore.createAgent({ name: "Non-immediate Agent", role: "executor" });
|
||||||
|
await agentStore.updateAgent(agent.id, {
|
||||||
|
runtimeConfig: { messageResponseMode: "on-heartbeat" },
|
||||||
|
});
|
||||||
|
|
||||||
|
const heartbeatMonitor = {
|
||||||
|
executeHeartbeat: vi.fn().mockResolvedValue({ id: "run-1" }),
|
||||||
|
};
|
||||||
|
|
||||||
|
const steeredTask = {
|
||||||
|
...FAKE_TASK_DETAIL,
|
||||||
|
id: "KB-001",
|
||||||
|
assignedAgentId: agent.id,
|
||||||
|
steeringComments: [{ id: "steer-1", text: "Please handle edge case", author: "user", createdAt: "2026-01-01T00:00:00.000Z" }],
|
||||||
|
};
|
||||||
|
(store.addSteeringComment as ReturnType<typeof vi.fn>).mockResolvedValue(steeredTask);
|
||||||
|
(store.getFusionDir as any) = vi.fn().mockReturnValue(fusionDir);
|
||||||
|
|
||||||
|
const app = express();
|
||||||
|
app.use(express.json());
|
||||||
|
app.use("/api", createApiRoutes(store, { heartbeatMonitor } as any));
|
||||||
|
|
||||||
|
const res = await REQUEST(
|
||||||
|
app,
|
||||||
|
"POST",
|
||||||
|
"/api/tasks/KB-001/steer",
|
||||||
|
JSON.stringify({ text: "Please handle edge case" }),
|
||||||
|
{ "Content-Type": "application/json" },
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(res.status).toBe(200);
|
||||||
|
await new Promise((resolve) => setTimeout(resolve, 0));
|
||||||
|
expect(heartbeatMonitor.executeHeartbeat).not.toHaveBeenCalled();
|
||||||
|
} finally {
|
||||||
|
rmSync(tempDir, { recursive: true, force: true });
|
||||||
|
}
|
||||||
|
});
|
||||||
|
|
||||||
it("returns 400 when text is missing", async () => {
|
it("returns 400 when text is missing", async () => {
|
||||||
const res = await REQUEST(buildApp(), "POST", "/api/tasks/KB-001/steer", JSON.stringify({}), {
|
const res = await REQUEST(buildApp(), "POST", "/api/tasks/KB-001/steer", JSON.stringify({}), {
|
||||||
"Content-Type": "application/json",
|
"Content-Type": "application/json",
|
||||||
@@ -9324,6 +9577,70 @@ describe("POST /api/agents/:id/runs", () => {
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("accepts triggering comment wake fields and persists them in contextSnapshot", async () => {
|
||||||
|
const res = await REQUEST(
|
||||||
|
buildApp(),
|
||||||
|
"POST",
|
||||||
|
`/api/agents/${agentId}/runs`,
|
||||||
|
JSON.stringify({
|
||||||
|
source: "on_demand",
|
||||||
|
triggerDetail: "task-comment",
|
||||||
|
taskId: "FN-001",
|
||||||
|
triggeringCommentIds: ["c1", "c2"],
|
||||||
|
triggeringCommentType: "task",
|
||||||
|
}),
|
||||||
|
{ "Content-Type": "application/json" },
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(res.status).toBe(201);
|
||||||
|
expect(res.body.contextSnapshot).toMatchObject({
|
||||||
|
wakeReason: "on_demand",
|
||||||
|
triggerDetail: "task-comment",
|
||||||
|
taskId: "FN-001",
|
||||||
|
triggeringCommentIds: ["c1", "c2"],
|
||||||
|
triggeringCommentType: "task",
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
|
it("returns 400 when triggeringCommentIds is not an array", async () => {
|
||||||
|
const res = await REQUEST(
|
||||||
|
buildApp(),
|
||||||
|
"POST",
|
||||||
|
`/api/agents/${agentId}/runs`,
|
||||||
|
JSON.stringify({ triggeringCommentIds: "not-an-array" }),
|
||||||
|
{ "Content-Type": "application/json" },
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(res.status).toBe(400);
|
||||||
|
expect(res.body.error).toContain("triggeringCommentIds must be an array of strings");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("returns 400 when triggeringCommentIds contains non-string values", async () => {
|
||||||
|
const res = await REQUEST(
|
||||||
|
buildApp(),
|
||||||
|
"POST",
|
||||||
|
`/api/agents/${agentId}/runs`,
|
||||||
|
JSON.stringify({ triggeringCommentIds: ["c1", 42] }),
|
||||||
|
{ "Content-Type": "application/json" },
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(res.status).toBe(400);
|
||||||
|
expect(res.body.error).toContain("triggeringCommentIds must be an array of strings");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("returns 400 when triggeringCommentType is invalid", async () => {
|
||||||
|
const res = await REQUEST(
|
||||||
|
buildApp(),
|
||||||
|
"POST",
|
||||||
|
`/api/agents/${agentId}/runs`,
|
||||||
|
JSON.stringify({ triggeringCommentType: "invalid" }),
|
||||||
|
{ "Content-Type": "application/json" },
|
||||||
|
);
|
||||||
|
|
||||||
|
expect(res.status).toBe(400);
|
||||||
|
expect(res.body.error).toContain("triggeringCommentType must be one of: steering, task, pr");
|
||||||
|
});
|
||||||
|
|
||||||
it("includes wake context without taskId when not provided", async () => {
|
it("includes wake context without taskId when not provided", async () => {
|
||||||
const res = await REQUEST(
|
const res = await REQUEST(
|
||||||
buildApp(),
|
buildApp(),
|
||||||
|
|||||||
@@ -1443,6 +1443,58 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
|||||||
const hasHeartbeatExecutor = Boolean(heartbeatMonitor);
|
const hasHeartbeatExecutor = Boolean(heartbeatMonitor);
|
||||||
const aiSessionStore = options?.aiSessionStore;
|
const aiSessionStore = options?.aiSessionStore;
|
||||||
|
|
||||||
|
const triggerCommentWakeForAssignedAgent = async (
|
||||||
|
scopedStore: TaskStore,
|
||||||
|
task: Task,
|
||||||
|
wake: {
|
||||||
|
triggeringCommentType: "steering" | "task" | "pr";
|
||||||
|
triggeringCommentIds?: string[];
|
||||||
|
triggerDetail: string;
|
||||||
|
},
|
||||||
|
): Promise<void> => {
|
||||||
|
if (!hasHeartbeatExecutor || !heartbeatMonitor || !task.assignedAgentId) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
const { AgentStore } = await import("@fusion/core");
|
||||||
|
const agentStore = new AgentStore({ rootDir: scopedStore.getFusionDir() });
|
||||||
|
await agentStore.init();
|
||||||
|
|
||||||
|
const assignedAgent = await agentStore.getAgent(task.assignedAgentId);
|
||||||
|
if (!assignedAgent) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
const responseMode = (assignedAgent.runtimeConfig as { messageResponseMode?: string } | undefined)?.messageResponseMode;
|
||||||
|
if (responseMode !== "immediate") {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
const activeRun = await agentStore.getActiveHeartbeatRun(assignedAgent.id);
|
||||||
|
if (activeRun) {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
const triggeringCommentIds = wake.triggeringCommentIds?.filter((id) => typeof id === "string" && id.length > 0);
|
||||||
|
const contextSnapshot: Record<string, unknown> = {
|
||||||
|
wakeReason: "on_demand",
|
||||||
|
triggerDetail: wake.triggerDetail,
|
||||||
|
taskId: task.id,
|
||||||
|
...(triggeringCommentIds?.length ? { triggeringCommentIds } : {}),
|
||||||
|
triggeringCommentType: wake.triggeringCommentType,
|
||||||
|
};
|
||||||
|
|
||||||
|
await heartbeatMonitor.executeHeartbeat({
|
||||||
|
agentId: assignedAgent.id,
|
||||||
|
source: "on_demand",
|
||||||
|
triggerDetail: wake.triggerDetail,
|
||||||
|
taskId: task.id,
|
||||||
|
triggeringCommentIds,
|
||||||
|
triggeringCommentType: wake.triggeringCommentType,
|
||||||
|
contextSnapshot,
|
||||||
|
});
|
||||||
|
};
|
||||||
|
|
||||||
// Scheduler config (includes persisted settings)
|
// Scheduler config (includes persisted settings)
|
||||||
router.get("/config", async (req, res) => {
|
router.get("/config", async (req, res) => {
|
||||||
try {
|
try {
|
||||||
@@ -2644,6 +2696,18 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
|||||||
throw badRequest("author must be a string");
|
throw badRequest("author must be a string");
|
||||||
}
|
}
|
||||||
const task = await scopedStore.addTaskComment(req.params.id, text, author?.trim() || "user");
|
const task = await scopedStore.addTaskComment(req.params.id, text, author?.trim() || "user");
|
||||||
|
|
||||||
|
const newCommentId = task.comments?.at(-1)?.id;
|
||||||
|
void triggerCommentWakeForAssignedAgent(scopedStore, task, {
|
||||||
|
triggeringCommentType: "task",
|
||||||
|
triggeringCommentIds: newCommentId ? [newCommentId] : undefined,
|
||||||
|
triggerDetail: "task-comment",
|
||||||
|
}).catch((error) => {
|
||||||
|
console.warn(
|
||||||
|
`[routes] failed to trigger task-comment heartbeat for ${task.id}: ${error instanceof Error ? error.message : String(error)}`,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
res.json(task);
|
res.json(task);
|
||||||
} catch (err: any) {
|
} catch (err: any) {
|
||||||
if (err instanceof ApiError) {
|
if (err instanceof ApiError) {
|
||||||
@@ -2705,6 +2769,18 @@ export function createApiRoutes(store: TaskStore, options?: ServerOptions): Rout
|
|||||||
throw badRequest("text must be between 1 and 2000 characters");
|
throw badRequest("text must be between 1 and 2000 characters");
|
||||||
}
|
}
|
||||||
const task = await scopedStore.addSteeringComment(req.params.id, text, "user");
|
const task = await scopedStore.addSteeringComment(req.params.id, text, "user");
|
||||||
|
|
||||||
|
const newSteeringCommentId = task.steeringComments?.at(-1)?.id;
|
||||||
|
void triggerCommentWakeForAssignedAgent(scopedStore, task, {
|
||||||
|
triggeringCommentType: "steering",
|
||||||
|
triggeringCommentIds: newSteeringCommentId ? [newSteeringCommentId] : undefined,
|
||||||
|
triggerDetail: "steering-comment",
|
||||||
|
}).catch((error) => {
|
||||||
|
console.warn(
|
||||||
|
`[routes] failed to trigger steering-comment heartbeat for ${task.id}: ${error instanceof Error ? error.message : String(error)}`,
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
res.json(task);
|
res.json(task);
|
||||||
} catch (err: any) {
|
} catch (err: any) {
|
||||||
if (err instanceof ApiError) {
|
if (err instanceof ApiError) {
|
||||||
@@ -8828,7 +8904,13 @@ Output ONLY the prompt text (no markdown, no explanations).`;
|
|||||||
/**
|
/**
|
||||||
* POST /api/agents/:id/runs
|
* POST /api/agents/:id/runs
|
||||||
* Manually start a heartbeat run for an agent.
|
* Manually start a heartbeat run for an agent.
|
||||||
* Body: { source?: HeartbeatInvocationSource, triggerDetail?: string, taskId?: string }
|
* Body: {
|
||||||
|
* source?: HeartbeatInvocationSource,
|
||||||
|
* triggerDetail?: string,
|
||||||
|
* taskId?: string,
|
||||||
|
* triggeringCommentIds?: string[],
|
||||||
|
* triggeringCommentType?: "steering" | "task" | "pr",
|
||||||
|
* }
|
||||||
*
|
*
|
||||||
* When HeartbeatMonitor is available, delegates to executeHeartbeat() with
|
* When HeartbeatMonitor is available, delegates to executeHeartbeat() with
|
||||||
* a structured wake context snapshot. This ensures a single authoritative run
|
* a structured wake context snapshot. This ensures a single authoritative run
|
||||||
@@ -8838,10 +8920,32 @@ Output ONLY the prompt text (no markdown, no explanations).`;
|
|||||||
*/
|
*/
|
||||||
router.post("/agents/:id/runs", async (req, res) => {
|
router.post("/agents/:id/runs", async (req, res) => {
|
||||||
try {
|
try {
|
||||||
const { source, triggerDetail, taskId } = req.body || {};
|
const { source, triggerDetail, taskId, triggeringCommentIds, triggeringCommentType } = req.body || {};
|
||||||
const invocationSource = source ?? "on_demand";
|
const invocationSource = source ?? "on_demand";
|
||||||
const trigger = triggerDetail ?? "Triggered from dashboard";
|
const trigger = triggerDetail ?? "Triggered from dashboard";
|
||||||
|
|
||||||
|
if (triggeringCommentIds !== undefined) {
|
||||||
|
if (!Array.isArray(triggeringCommentIds) || triggeringCommentIds.some((id) => typeof id !== "string")) {
|
||||||
|
throw badRequest("triggeringCommentIds must be an array of strings");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
if (
|
||||||
|
triggeringCommentType !== undefined
|
||||||
|
&& triggeringCommentType !== "steering"
|
||||||
|
&& triggeringCommentType !== "task"
|
||||||
|
&& triggeringCommentType !== "pr"
|
||||||
|
) {
|
||||||
|
throw badRequest("triggeringCommentType must be one of: steering, task, pr");
|
||||||
|
}
|
||||||
|
|
||||||
|
const normalizedTriggeringCommentIds = Array.isArray(triggeringCommentIds)
|
||||||
|
? triggeringCommentIds.map((id) => id.trim()).filter((id) => id.length > 0)
|
||||||
|
: undefined;
|
||||||
|
const normalizedTriggeringCommentType =
|
||||||
|
triggeringCommentType === "steering" || triggeringCommentType === "task" || triggeringCommentType === "pr"
|
||||||
|
? triggeringCommentType
|
||||||
|
: undefined;
|
||||||
|
|
||||||
// Build structured wake context
|
// Build structured wake context
|
||||||
const contextSnapshot: Record<string, unknown> = {
|
const contextSnapshot: Record<string, unknown> = {
|
||||||
wakeReason: invocationSource,
|
wakeReason: invocationSource,
|
||||||
@@ -8850,6 +8954,12 @@ Output ONLY the prompt text (no markdown, no explanations).`;
|
|||||||
if (taskId) {
|
if (taskId) {
|
||||||
contextSnapshot.taskId = taskId;
|
contextSnapshot.taskId = taskId;
|
||||||
}
|
}
|
||||||
|
if (normalizedTriggeringCommentIds?.length) {
|
||||||
|
contextSnapshot.triggeringCommentIds = normalizedTriggeringCommentIds;
|
||||||
|
}
|
||||||
|
if (normalizedTriggeringCommentType) {
|
||||||
|
contextSnapshot.triggeringCommentType = normalizedTriggeringCommentType;
|
||||||
|
}
|
||||||
|
|
||||||
if (hasHeartbeatExecutor && heartbeatMonitor) {
|
if (hasHeartbeatExecutor && heartbeatMonitor) {
|
||||||
// Check for existing active run
|
// Check for existing active run
|
||||||
@@ -8869,6 +8979,8 @@ Output ONLY the prompt text (no markdown, no explanations).`;
|
|||||||
source: invocationSource,
|
source: invocationSource,
|
||||||
triggerDetail: trigger,
|
triggerDetail: trigger,
|
||||||
taskId,
|
taskId,
|
||||||
|
triggeringCommentIds: normalizedTriggeringCommentIds,
|
||||||
|
triggeringCommentType: normalizedTriggeringCommentType,
|
||||||
contextSnapshot,
|
contextSnapshot,
|
||||||
});
|
});
|
||||||
|
|
||||||
|
|||||||
@@ -92,7 +92,15 @@ export interface ServerOptions {
|
|||||||
/** Optional HeartbeatMonitor for triggering agent execution runs */
|
/** Optional HeartbeatMonitor for triggering agent execution runs */
|
||||||
heartbeatMonitor?: {
|
heartbeatMonitor?: {
|
||||||
startRun(agentId: string, options?: { source: import("@fusion/core").HeartbeatInvocationSource; triggerDetail?: string; contextSnapshot?: Record<string, unknown> }): Promise<import("@fusion/core").AgentHeartbeatRun>;
|
startRun(agentId: string, options?: { source: import("@fusion/core").HeartbeatInvocationSource; triggerDetail?: string; contextSnapshot?: Record<string, unknown> }): Promise<import("@fusion/core").AgentHeartbeatRun>;
|
||||||
executeHeartbeat(options: { agentId: string; source: import("@fusion/core").HeartbeatInvocationSource; triggerDetail?: string; taskId?: string; contextSnapshot?: Record<string, unknown> }): Promise<import("@fusion/core").AgentHeartbeatRun>;
|
executeHeartbeat(options: {
|
||||||
|
agentId: string;
|
||||||
|
source: import("@fusion/core").HeartbeatInvocationSource;
|
||||||
|
triggerDetail?: string;
|
||||||
|
taskId?: string;
|
||||||
|
triggeringCommentIds?: string[];
|
||||||
|
triggeringCommentType?: "steering" | "task" | "pr";
|
||||||
|
contextSnapshot?: Record<string, unknown>;
|
||||||
|
}): Promise<import("@fusion/core").AgentHeartbeatRun>;
|
||||||
stopRun(agentId: string): Promise<void>;
|
stopRun(agentId: string): Promise<void>;
|
||||||
};
|
};
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -1,5 +1,5 @@
|
|||||||
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
import { describe, it, expect, vi, beforeEach, afterEach } from "vitest";
|
||||||
import { HeartbeatMonitor, HeartbeatTriggerScheduler, type AgentSession, type HeartbeatExecutionOptions, HEARTBEAT_SYSTEM_PROMPT } from "./agent-heartbeat.js";
|
import { HeartbeatMonitor, HeartbeatTriggerScheduler, isBlockedStateDuplicate, type AgentSession, type HeartbeatExecutionOptions, HEARTBEAT_SYSTEM_PROMPT } from "./agent-heartbeat.js";
|
||||||
import { AgentLogger } from "./agent-logger.js";
|
import { AgentLogger } from "./agent-logger.js";
|
||||||
import type { AgentStore, AgentHeartbeatRun, TaskStore, TaskDetail, Agent, MessageStore, Message, AgentBudgetStatus } from "@fusion/core";
|
import type { AgentStore, AgentHeartbeatRun, TaskStore, TaskDetail, Agent, MessageStore, Message, AgentBudgetStatus } from "@fusion/core";
|
||||||
|
|
||||||
@@ -130,6 +130,32 @@ describe("HeartbeatMonitor", () => {
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("isBlockedStateDuplicate", () => {
|
||||||
|
it("returns true when blockedBy and contextHash match", () => {
|
||||||
|
expect(
|
||||||
|
isBlockedStateDuplicate(
|
||||||
|
{ taskId: "FN-1", blockedBy: "FN-0", recordedAt: "2026-01-01T00:00:00.000Z", contextHash: "abc" },
|
||||||
|
{ taskId: "FN-1", blockedBy: "FN-0", recordedAt: "2026-01-02T00:00:00.000Z", contextHash: "abc" },
|
||||||
|
),
|
||||||
|
).toBe(true);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("returns false when blockedBy differs or contextHash differs", () => {
|
||||||
|
expect(
|
||||||
|
isBlockedStateDuplicate(
|
||||||
|
{ taskId: "FN-1", blockedBy: "FN-0", recordedAt: "2026-01-01T00:00:00.000Z", contextHash: "abc" },
|
||||||
|
{ taskId: "FN-1", blockedBy: "FN-2", recordedAt: "2026-01-02T00:00:00.000Z", contextHash: "abc" },
|
||||||
|
),
|
||||||
|
).toBe(false);
|
||||||
|
expect(
|
||||||
|
isBlockedStateDuplicate(
|
||||||
|
{ taskId: "FN-1", blockedBy: "FN-0", recordedAt: "2026-01-01T00:00:00.000Z", contextHash: "abc" },
|
||||||
|
{ taskId: "FN-1", blockedBy: "FN-0", recordedAt: "2026-01-02T00:00:00.000Z", contextHash: "xyz" },
|
||||||
|
),
|
||||||
|
).toBe(false);
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe("start", () => {
|
describe("start", () => {
|
||||||
it("initiates polling interval", () => {
|
it("initiates polling interval", () => {
|
||||||
vi.useFakeTimers({ shouldAdvanceTime: true });
|
vi.useFakeTimers({ shouldAdvanceTime: true });
|
||||||
@@ -988,6 +1014,7 @@ describe("HeartbeatMonitor", () => {
|
|||||||
column: "triage",
|
column: "triage",
|
||||||
}),
|
}),
|
||||||
logEntry: vi.fn().mockResolvedValue({}),
|
logEntry: vi.fn().mockResolvedValue({}),
|
||||||
|
addComment: vi.fn().mockResolvedValue({}),
|
||||||
appendAgentLog: vi.fn().mockResolvedValue(undefined),
|
appendAgentLog: vi.fn().mockResolvedValue(undefined),
|
||||||
...overrides,
|
...overrides,
|
||||||
} as unknown as TaskStore;
|
} as unknown as TaskStore;
|
||||||
@@ -1041,6 +1068,9 @@ describe("HeartbeatMonitor", () => {
|
|||||||
endHeartbeatRun: vi.fn().mockResolvedValue(undefined),
|
endHeartbeatRun: vi.fn().mockResolvedValue(undefined),
|
||||||
getBudgetStatus: vi.fn().mockResolvedValue(createBudgetStatus()),
|
getBudgetStatus: vi.fn().mockResolvedValue(createBudgetStatus()),
|
||||||
getCachedAgent: vi.fn().mockReturnValue(null),
|
getCachedAgent: vi.fn().mockReturnValue(null),
|
||||||
|
getLastBlockedState: vi.fn().mockResolvedValue(null),
|
||||||
|
setLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
||||||
|
clearLastBlockedState: vi.fn().mockResolvedValue(undefined),
|
||||||
} as unknown as AgentStore;
|
} as unknown as AgentStore;
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -1125,6 +1155,171 @@ describe("HeartbeatMonitor", () => {
|
|||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
|
describe("blocked-task dedup", () => {
|
||||||
|
const buildContextHash = (blockedBy: string, taskDetail: Partial<TaskDetail>): string => {
|
||||||
|
const commentCount = (taskDetail.comments?.length ?? 0) + (taskDetail.steeringComments?.length ?? 0);
|
||||||
|
const lastCommentId = taskDetail.comments?.at(-1)?.id;
|
||||||
|
const lastSteeringCommentId = taskDetail.steeringComments?.at(-1)?.id;
|
||||||
|
|
||||||
|
return Buffer.from(
|
||||||
|
JSON.stringify({ commentCount, lastCommentId, lastSteeringCommentId, blockedBy }),
|
||||||
|
)
|
||||||
|
.toString("base64")
|
||||||
|
.slice(0, 16);
|
||||||
|
};
|
||||||
|
|
||||||
|
it("skips duplicate blocked comments when blocked snapshot is unchanged", async () => {
|
||||||
|
const store = createStoreWithAgentForExec({ taskId: "FN-BLOCKED" });
|
||||||
|
const taskDetail = {
|
||||||
|
id: "FN-BLOCKED",
|
||||||
|
title: "Blocked Task",
|
||||||
|
description: "Blocked task description",
|
||||||
|
prompt: "",
|
||||||
|
status: "queued",
|
||||||
|
blockedBy: "FN-DEP-1",
|
||||||
|
comments: [{ id: "comment-1", text: "Still blocked", author: "user", createdAt: "2026-01-01T00:00:00.000Z" }],
|
||||||
|
steeringComments: [],
|
||||||
|
steps: [],
|
||||||
|
column: "todo",
|
||||||
|
dependencies: [],
|
||||||
|
log: [],
|
||||||
|
attachments: [],
|
||||||
|
createdAt: new Date().toISOString(),
|
||||||
|
updatedAt: new Date().toISOString(),
|
||||||
|
} as unknown as TaskDetail;
|
||||||
|
|
||||||
|
(store.getLastBlockedState as ReturnType<typeof vi.fn>).mockResolvedValue({
|
||||||
|
taskId: "FN-BLOCKED",
|
||||||
|
blockedBy: "FN-DEP-1",
|
||||||
|
recordedAt: "2026-01-01T00:00:00.000Z",
|
||||||
|
contextHash: buildContextHash("FN-DEP-1", taskDetail),
|
||||||
|
});
|
||||||
|
|
||||||
|
mockTaskStore = createMockTaskStore({
|
||||||
|
getTask: vi.fn().mockResolvedValue(taskDetail),
|
||||||
|
});
|
||||||
|
|
||||||
|
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||||
|
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "timer" });
|
||||||
|
|
||||||
|
expect(result.resultJson).toEqual({ reason: "blocked_duplicate", taskId: "FN-BLOCKED", blockedBy: "FN-DEP-1" });
|
||||||
|
expect(mockTaskStore.addComment).not.toHaveBeenCalled();
|
||||||
|
expect(store.setLastBlockedState).not.toHaveBeenCalled();
|
||||||
|
expect(mockedCreateKbAgent).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("re-logs blocked state when new comments change context hash", async () => {
|
||||||
|
const store = createStoreWithAgentForExec({ taskId: "FN-BLOCKED" });
|
||||||
|
(store.getLastBlockedState as ReturnType<typeof vi.fn>).mockResolvedValue({
|
||||||
|
taskId: "FN-BLOCKED",
|
||||||
|
blockedBy: "FN-DEP-1",
|
||||||
|
recordedAt: "2026-01-01T00:00:00.000Z",
|
||||||
|
contextHash: "stale-context-hash",
|
||||||
|
});
|
||||||
|
|
||||||
|
const taskDetail = {
|
||||||
|
id: "FN-BLOCKED",
|
||||||
|
title: "Blocked Task",
|
||||||
|
description: "Blocked task description",
|
||||||
|
prompt: "",
|
||||||
|
status: "queued",
|
||||||
|
blockedBy: "FN-DEP-1",
|
||||||
|
comments: [{ id: "comment-2", text: "New context", author: "user", createdAt: "2026-01-02T00:00:00.000Z" }],
|
||||||
|
steeringComments: [],
|
||||||
|
steps: [],
|
||||||
|
column: "todo",
|
||||||
|
dependencies: [],
|
||||||
|
log: [],
|
||||||
|
attachments: [],
|
||||||
|
createdAt: new Date().toISOString(),
|
||||||
|
updatedAt: new Date().toISOString(),
|
||||||
|
} as unknown as TaskDetail;
|
||||||
|
|
||||||
|
mockTaskStore = createMockTaskStore({ getTask: vi.fn().mockResolvedValue(taskDetail) });
|
||||||
|
|
||||||
|
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||||
|
const result = await monitor.executeHeartbeat({ agentId: "agent-001", source: "timer" });
|
||||||
|
|
||||||
|
expect(result.resultJson).toEqual({ reason: "blocked", taskId: "FN-BLOCKED", blockedBy: "FN-DEP-1" });
|
||||||
|
expect(mockTaskStore.addComment).toHaveBeenCalledOnce();
|
||||||
|
expect(store.setLastBlockedState).toHaveBeenCalledWith(
|
||||||
|
"agent-001",
|
||||||
|
expect.objectContaining({ taskId: "FN-BLOCKED", blockedBy: "FN-DEP-1" }),
|
||||||
|
);
|
||||||
|
expect(mockedCreateKbAgent).not.toHaveBeenCalled();
|
||||||
|
});
|
||||||
|
|
||||||
|
it("treats changed blockedBy as a new blocked state", async () => {
|
||||||
|
const store = createStoreWithAgentForExec({ taskId: "FN-BLOCKED" });
|
||||||
|
(store.getLastBlockedState as ReturnType<typeof vi.fn>).mockResolvedValue({
|
||||||
|
taskId: "FN-BLOCKED",
|
||||||
|
blockedBy: "FN-DEP-OLD",
|
||||||
|
recordedAt: "2026-01-01T00:00:00.000Z",
|
||||||
|
contextHash: "samehash",
|
||||||
|
});
|
||||||
|
|
||||||
|
const taskDetail = {
|
||||||
|
id: "FN-BLOCKED",
|
||||||
|
title: "Blocked Task",
|
||||||
|
description: "Blocked task description",
|
||||||
|
prompt: "",
|
||||||
|
status: "queued",
|
||||||
|
blockedBy: "FN-DEP-NEW",
|
||||||
|
comments: [],
|
||||||
|
steeringComments: [],
|
||||||
|
steps: [],
|
||||||
|
column: "todo",
|
||||||
|
dependencies: [],
|
||||||
|
log: [],
|
||||||
|
attachments: [],
|
||||||
|
createdAt: new Date().toISOString(),
|
||||||
|
updatedAt: new Date().toISOString(),
|
||||||
|
} as unknown as TaskDetail;
|
||||||
|
|
||||||
|
mockTaskStore = createMockTaskStore({ getTask: vi.fn().mockResolvedValue(taskDetail) });
|
||||||
|
|
||||||
|
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||||
|
await monitor.executeHeartbeat({ agentId: "agent-001", source: "timer" });
|
||||||
|
|
||||||
|
expect(mockTaskStore.addComment).toHaveBeenCalledOnce();
|
||||||
|
expect(store.setLastBlockedState).toHaveBeenCalledWith(
|
||||||
|
"agent-001",
|
||||||
|
expect.objectContaining({ blockedBy: "FN-DEP-NEW" }),
|
||||||
|
);
|
||||||
|
});
|
||||||
|
|
||||||
|
it("clears blocked state when task is no longer blocked", async () => {
|
||||||
|
const store = createStoreWithAgentForExec({ taskId: "FN-READY" });
|
||||||
|
const mockSession = createMockAgentSession();
|
||||||
|
mockedCreateKbAgent.mockResolvedValue({ session: mockSession as any });
|
||||||
|
|
||||||
|
mockTaskStore = createMockTaskStore({
|
||||||
|
getTask: vi.fn().mockResolvedValue({
|
||||||
|
id: "FN-READY",
|
||||||
|
title: "Ready Task",
|
||||||
|
description: "Ready to run",
|
||||||
|
prompt: "",
|
||||||
|
status: undefined,
|
||||||
|
blockedBy: undefined,
|
||||||
|
comments: [],
|
||||||
|
steeringComments: [],
|
||||||
|
steps: [],
|
||||||
|
column: "todo",
|
||||||
|
dependencies: [],
|
||||||
|
log: [],
|
||||||
|
attachments: [],
|
||||||
|
createdAt: new Date().toISOString(),
|
||||||
|
updatedAt: new Date().toISOString(),
|
||||||
|
} as unknown as TaskDetail),
|
||||||
|
});
|
||||||
|
|
||||||
|
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||||
|
await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
|
||||||
|
|
||||||
|
expect(store.clearLastBlockedState).toHaveBeenCalledWith("agent-001");
|
||||||
|
});
|
||||||
|
});
|
||||||
|
|
||||||
describe("executeHeartbeat - inbox selection", () => {
|
describe("executeHeartbeat - inbox selection", () => {
|
||||||
const makeInboxSelection = (taskId: string, priority: "in_progress" | "todo" | "blocked" = "todo") => {
|
const makeInboxSelection = (taskId: string, priority: "in_progress" | "todo" | "blocked" = "todo") => {
|
||||||
const now = new Date().toISOString();
|
const now = new Date().toISOString();
|
||||||
@@ -1349,6 +1544,59 @@ describe("HeartbeatMonitor", () => {
|
|||||||
expect(promptArg).toContain("PROMPT.md");
|
expect(promptArg).toContain("PROMPT.md");
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("includes triggering comment context in execution prompt when comment IDs are provided", async () => {
|
||||||
|
const store = createStoreWithAgentForExec();
|
||||||
|
const mockSession = createMockAgentSession();
|
||||||
|
mockedCreateKbAgent.mockResolvedValue({
|
||||||
|
session: mockSession as any,
|
||||||
|
});
|
||||||
|
|
||||||
|
mockTaskStore.getTask = vi.fn().mockResolvedValue({
|
||||||
|
id: "FN-001",
|
||||||
|
title: "Test Task",
|
||||||
|
description: "Test task description",
|
||||||
|
prompt: "# Prompt",
|
||||||
|
comments: [{ id: "c-1", author: "user", text: "Please cover edge cases", createdAt: "2026-01-01T00:00:00.000Z" }],
|
||||||
|
steeringComments: [{ id: "s-1", author: "agent", text: "Investigating blocker", createdAt: "2026-01-01T00:01:00.000Z" }],
|
||||||
|
steps: [],
|
||||||
|
column: "todo",
|
||||||
|
dependencies: [],
|
||||||
|
log: [],
|
||||||
|
attachments: [],
|
||||||
|
createdAt: new Date().toISOString(),
|
||||||
|
updatedAt: new Date().toISOString(),
|
||||||
|
} as unknown as TaskDetail);
|
||||||
|
|
||||||
|
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||||
|
|
||||||
|
await monitor.executeHeartbeat({
|
||||||
|
agentId: "agent-001",
|
||||||
|
source: "on_demand",
|
||||||
|
triggeringCommentIds: ["c-1", "s-1"],
|
||||||
|
triggeringCommentType: "steering",
|
||||||
|
});
|
||||||
|
|
||||||
|
const promptArg = mockSession.prompt.mock.calls[0]![0] as string;
|
||||||
|
expect(promptArg).toContain("You were woken because of new comments on this task");
|
||||||
|
expect(promptArg).toContain("Please cover edge cases");
|
||||||
|
expect(promptArg).toContain("Investigating blocker");
|
||||||
|
});
|
||||||
|
|
||||||
|
it("keeps standard prompt when no triggering comments are provided", async () => {
|
||||||
|
const store = createStoreWithAgentForExec();
|
||||||
|
const mockSession = createMockAgentSession();
|
||||||
|
mockedCreateKbAgent.mockResolvedValue({
|
||||||
|
session: mockSession as any,
|
||||||
|
});
|
||||||
|
|
||||||
|
const monitor = new HeartbeatMonitor({ store, taskStore: mockTaskStore, rootDir: "/tmp" });
|
||||||
|
await monitor.executeHeartbeat({ agentId: "agent-001", source: "on_demand" });
|
||||||
|
|
||||||
|
const promptArg = mockSession.prompt.mock.calls[0]![0] as string;
|
||||||
|
expect(promptArg).not.toContain("You were woken because of new comments on this task");
|
||||||
|
expect(promptArg).not.toContain("New comments since last run:");
|
||||||
|
});
|
||||||
|
|
||||||
it("completes run with status completed on successful execution", async () => {
|
it("completes run with status completed on successful execution", async () => {
|
||||||
const store = createStoreWithAgentForExec();
|
const store = createStoreWithAgentForExec();
|
||||||
const mockSession = createMockAgentSession();
|
const mockSession = createMockAgentSession();
|
||||||
@@ -1454,6 +1702,8 @@ describe("HeartbeatMonitor", () => {
|
|||||||
agentId: "agent-001",
|
agentId: "agent-001",
|
||||||
source: "assignment",
|
source: "assignment",
|
||||||
triggerDetail: "task-assigned",
|
triggerDetail: "task-assigned",
|
||||||
|
triggeringCommentIds: ["comment-1"],
|
||||||
|
triggeringCommentType: "task",
|
||||||
contextSnapshot: {
|
contextSnapshot: {
|
||||||
wakeReason: "assignment",
|
wakeReason: "assignment",
|
||||||
triggerDetail: "task-assigned",
|
triggerDetail: "task-assigned",
|
||||||
@@ -1465,6 +1715,8 @@ describe("HeartbeatMonitor", () => {
|
|||||||
wakeReason: "assignment",
|
wakeReason: "assignment",
|
||||||
triggerDetail: "task-assigned",
|
triggerDetail: "task-assigned",
|
||||||
taskId: "FN-001",
|
taskId: "FN-001",
|
||||||
|
triggeringCommentIds: ["comment-1"],
|
||||||
|
triggeringCommentType: "task",
|
||||||
});
|
});
|
||||||
});
|
});
|
||||||
|
|
||||||
@@ -2734,6 +2986,40 @@ describe("HeartbeatTriggerScheduler", () => {
|
|||||||
);
|
);
|
||||||
});
|
});
|
||||||
|
|
||||||
|
it("includes new steering comment IDs for assignment wakes when taskStore is available", async () => {
|
||||||
|
scheduler.stop();
|
||||||
|
|
||||||
|
(eventStore as any).getRecentRuns = vi.fn().mockResolvedValue([
|
||||||
|
{ startedAt: "2026-01-01T00:00:00.000Z" },
|
||||||
|
]);
|
||||||
|
|
||||||
|
const assignmentTaskStore = {
|
||||||
|
getTask: vi.fn().mockResolvedValue({
|
||||||
|
id: "FN-006",
|
||||||
|
steeringComments: [
|
||||||
|
{ id: "steer-old", text: "older", author: "user", createdAt: "2025-12-31T23:00:00.000Z" },
|
||||||
|
{ id: "steer-new", text: "new guidance", author: "user", createdAt: "2026-01-01T01:00:00.000Z" },
|
||||||
|
],
|
||||||
|
}),
|
||||||
|
} as unknown as TaskStore;
|
||||||
|
|
||||||
|
scheduler = new HeartbeatTriggerScheduler(eventStore, callback, assignmentTaskStore);
|
||||||
|
scheduler.start();
|
||||||
|
|
||||||
|
const agent = { id: "agent-test", name: "Test" } as import("@fusion/core").Agent;
|
||||||
|
eventStore.emit("agent:assigned", agent, "FN-006");
|
||||||
|
|
||||||
|
await vi.waitFor(() => {
|
||||||
|
expect(callback).toHaveBeenCalledOnce();
|
||||||
|
}, { timeout: 1000 });
|
||||||
|
|
||||||
|
expect(callback).toHaveBeenCalledWith("agent-test", "assignment", expect.objectContaining({
|
||||||
|
taskId: "FN-006",
|
||||||
|
triggeringCommentIds: ["steer-new"],
|
||||||
|
triggeringCommentType: "steering",
|
||||||
|
}));
|
||||||
|
});
|
||||||
|
|
||||||
it("cleans up listener on unwatch", async () => {
|
it("cleans up listener on unwatch", async () => {
|
||||||
scheduler.unwatchAssignments();
|
scheduler.unwatchAssignments();
|
||||||
|
|
||||||
|
|||||||
@@ -17,7 +17,7 @@
|
|||||||
* - onTerminated: Called when an unresponsive agent is terminated
|
* - onTerminated: Called when an unresponsive agent is terminated
|
||||||
*/
|
*/
|
||||||
|
|
||||||
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent, InboxTask } from "@fusion/core";
|
import type { AgentStore, AgentHeartbeatRun, HeartbeatInvocationSource, AgentHeartbeatConfig, AgentBudgetStatus, Message, MessageStore, TaskStore, TaskDetail, AgentRole, Agent, InboxTask, BlockedStateSnapshot } from "@fusion/core";
|
||||||
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
|
import type { ToolDefinition } from "@mariozechner/pi-coding-agent";
|
||||||
import { Type, type Static } from "@mariozechner/pi-ai";
|
import { Type, type Static } from "@mariozechner/pi-ai";
|
||||||
import { createTaskCreateTool, createTaskLogTool, taskCreateParams } from "./agent-tools.js";
|
import { createTaskCreateTool, createTaskLogTool, taskCreateParams } from "./agent-tools.js";
|
||||||
@@ -89,6 +89,10 @@ export interface HeartbeatExecutionOptions {
|
|||||||
triggerDetail?: string;
|
triggerDetail?: string;
|
||||||
/** Optional task ID override (uses agent.taskId if not set) */
|
/** Optional task ID override (uses agent.taskId if not set) */
|
||||||
taskId?: string;
|
taskId?: string;
|
||||||
|
/** IDs of comments that triggered this wake (if any) */
|
||||||
|
triggeringCommentIds?: string[];
|
||||||
|
/** Type of comment that triggered this wake */
|
||||||
|
triggeringCommentType?: "steering" | "task" | "pr";
|
||||||
/** Optional structured context persisted on the run record */
|
/** Optional structured context persisted on the run record */
|
||||||
contextSnapshot?: Record<string, unknown>;
|
contextSnapshot?: Record<string, unknown>;
|
||||||
}
|
}
|
||||||
@@ -110,6 +114,11 @@ interface TrackedAgent {
|
|||||||
sessionIdBefore?: string;
|
sessionIdBefore?: string;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/** Compare blocked-state snapshots to decide whether blocked messaging is duplicate noise. */
|
||||||
|
export function isBlockedStateDuplicate(current: BlockedStateSnapshot, previous: BlockedStateSnapshot): boolean {
|
||||||
|
return current.blockedBy === previous.blockedBy && current.contextHash === previous.contextHash;
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* System prompt for heartbeat agent sessions.
|
* System prompt for heartbeat agent sessions.
|
||||||
* Instructs the agent to perform a single-pass check on its assigned task
|
* Instructs the agent to perform a single-pass check on its assigned task
|
||||||
@@ -573,7 +582,15 @@ export class HeartbeatMonitor {
|
|||||||
* @throws Error if taskStore or rootDir are not configured
|
* @throws Error if taskStore or rootDir are not configured
|
||||||
*/
|
*/
|
||||||
async executeHeartbeat(options: HeartbeatExecutionOptions): Promise<AgentHeartbeatRun> {
|
async executeHeartbeat(options: HeartbeatExecutionOptions): Promise<AgentHeartbeatRun> {
|
||||||
const { agentId, source, triggerDetail, taskId: explicitTaskId, contextSnapshot } = options;
|
const {
|
||||||
|
agentId,
|
||||||
|
source,
|
||||||
|
triggerDetail,
|
||||||
|
taskId: explicitTaskId,
|
||||||
|
contextSnapshot,
|
||||||
|
triggeringCommentIds,
|
||||||
|
triggeringCommentType,
|
||||||
|
} = options;
|
||||||
|
|
||||||
// Validate execution dependencies
|
// Validate execution dependencies
|
||||||
if (!this.taskStore || !this.rootDir) {
|
if (!this.taskStore || !this.rootDir) {
|
||||||
@@ -594,9 +611,25 @@ export class HeartbeatMonitor {
|
|||||||
}
|
}
|
||||||
|
|
||||||
const resolvedTaskId = explicitTaskId ?? preloadedAgent?.taskId;
|
const resolvedTaskId = explicitTaskId ?? preloadedAgent?.taskId;
|
||||||
|
const contextTriggeringCommentIds = Array.isArray(contextSnapshot?.triggeringCommentIds)
|
||||||
|
? contextSnapshot.triggeringCommentIds.filter((id): id is string => typeof id === "string" && id.length > 0)
|
||||||
|
: undefined;
|
||||||
|
const contextTriggeringCommentType =
|
||||||
|
contextSnapshot?.triggeringCommentType === "steering"
|
||||||
|
|| contextSnapshot?.triggeringCommentType === "task"
|
||||||
|
|| contextSnapshot?.triggeringCommentType === "pr"
|
||||||
|
? contextSnapshot.triggeringCommentType
|
||||||
|
: undefined;
|
||||||
|
const effectiveTriggeringCommentIds = triggeringCommentIds ?? contextTriggeringCommentIds;
|
||||||
|
const effectiveTriggeringCommentType = triggeringCommentType ?? contextTriggeringCommentType;
|
||||||
|
|
||||||
const runContextSnapshot = {
|
const runContextSnapshot = {
|
||||||
...(contextSnapshot ?? {}),
|
...(contextSnapshot ?? {}),
|
||||||
...(resolvedTaskId ? { taskId: resolvedTaskId } : {}),
|
...(resolvedTaskId ? { taskId: resolvedTaskId } : {}),
|
||||||
|
...(effectiveTriggeringCommentIds?.length
|
||||||
|
? { triggeringCommentIds: effectiveTriggeringCommentIds }
|
||||||
|
: {}),
|
||||||
|
...(effectiveTriggeringCommentType ? { triggeringCommentType: effectiveTriggeringCommentType } : {}),
|
||||||
};
|
};
|
||||||
|
|
||||||
// Start run
|
// Start run
|
||||||
@@ -752,6 +785,50 @@ export class HeartbeatMonitor {
|
|||||||
return (await this.store.getRunDetail(agentId, run.id))!;
|
return (await this.store.getRunDetail(agentId, run.id))!;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
const blockedBy = typeof taskDetail.blockedBy === "string" ? taskDetail.blockedBy.trim() : "";
|
||||||
|
const isBlockedTask = taskDetail.status === "queued" && blockedBy.length > 0;
|
||||||
|
|
||||||
|
if (isBlockedTask) {
|
||||||
|
const commentCount = (taskDetail.comments?.length ?? 0) + (taskDetail.steeringComments?.length ?? 0);
|
||||||
|
const lastCommentId = taskDetail.comments?.at(-1)?.id;
|
||||||
|
const lastSteeringCommentId = taskDetail.steeringComments?.at(-1)?.id;
|
||||||
|
const contextHash = Buffer.from(
|
||||||
|
JSON.stringify({ commentCount, lastCommentId, lastSteeringCommentId, blockedBy }),
|
||||||
|
)
|
||||||
|
.toString("base64")
|
||||||
|
.slice(0, 16);
|
||||||
|
|
||||||
|
const currentBlockedState: BlockedStateSnapshot = {
|
||||||
|
taskId,
|
||||||
|
blockedBy,
|
||||||
|
recordedAt: new Date().toISOString(),
|
||||||
|
contextHash,
|
||||||
|
};
|
||||||
|
|
||||||
|
const previousBlockedState = await this.store.getLastBlockedState(agentId);
|
||||||
|
if (previousBlockedState && isBlockedStateDuplicate(currentBlockedState, previousBlockedState)) {
|
||||||
|
heartbeatLog.log(`Task ${taskId} is still blocked by ${blockedBy} (duplicate state) — skipping comment`);
|
||||||
|
await this.completeRun(agentId, run.id, {
|
||||||
|
status: "completed",
|
||||||
|
resultJson: { reason: "blocked_duplicate", taskId, blockedBy },
|
||||||
|
});
|
||||||
|
return (await this.store.getRunDetail(agentId, run.id))!;
|
||||||
|
}
|
||||||
|
|
||||||
|
const blockedMessage = `Task is blocked by ${blockedBy}; waiting for dependency/context changes before retrying.`;
|
||||||
|
await taskStore.addComment(taskId, blockedMessage, "agent");
|
||||||
|
await this.store.setLastBlockedState(agentId, currentBlockedState);
|
||||||
|
|
||||||
|
heartbeatLog.log(`Task ${taskId} is blocked by ${blockedBy} — recorded blocked state`);
|
||||||
|
await this.completeRun(agentId, run.id, {
|
||||||
|
status: "completed",
|
||||||
|
resultJson: { reason: "blocked", taskId, blockedBy },
|
||||||
|
});
|
||||||
|
return (await this.store.getRunDetail(agentId, run.id))!;
|
||||||
|
}
|
||||||
|
|
||||||
|
await this.store.clearLastBlockedState(agentId);
|
||||||
|
|
||||||
// Track usage via callbacks
|
// Track usage via callbacks
|
||||||
const STDOUT_EXCERPT_LIMIT = 4000;
|
const STDOUT_EXCERPT_LIMIT = 4000;
|
||||||
let outputLength = 0;
|
let outputLength = 0;
|
||||||
@@ -831,6 +908,36 @@ export class HeartbeatMonitor {
|
|||||||
try {
|
try {
|
||||||
// Build execution prompt
|
// Build execution prompt
|
||||||
const taskTitle = taskDetail.title ?? taskDetail.description.slice(0, 100);
|
const taskTitle = taskDetail.title ?? taskDetail.description.slice(0, 100);
|
||||||
|
|
||||||
|
const triggeringCommentLines: string[] = [];
|
||||||
|
if (effectiveTriggeringCommentIds && effectiveTriggeringCommentIds.length > 0) {
|
||||||
|
const commentLookup = new Map<string, { author: string; text: string }>();
|
||||||
|
for (const comment of taskDetail.comments ?? []) {
|
||||||
|
commentLookup.set(comment.id, { author: comment.author, text: comment.text });
|
||||||
|
}
|
||||||
|
for (const steeringComment of taskDetail.steeringComments ?? []) {
|
||||||
|
commentLookup.set(steeringComment.id, { author: steeringComment.author, text: steeringComment.text });
|
||||||
|
}
|
||||||
|
|
||||||
|
const formatCommentText = (text: string): string => text.replace(/\s+/g, " ").trim();
|
||||||
|
|
||||||
|
for (const commentId of effectiveTriggeringCommentIds) {
|
||||||
|
const comment = commentLookup.get(commentId);
|
||||||
|
if (comment) {
|
||||||
|
triggeringCommentLines.push(`- [${comment.author}]: "${formatCommentText(comment.text)}"`);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if (triggeringCommentLines.length > 0) {
|
||||||
|
triggeringCommentLines.unshift(
|
||||||
|
"",
|
||||||
|
"You were woken because of new comments on this task. Review them and take appropriate action.",
|
||||||
|
`Triggering comment type: ${effectiveTriggeringCommentType ?? "task"}`,
|
||||||
|
"New comments since last run:",
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
const executionPrompt = [
|
const executionPrompt = [
|
||||||
`Heartbeat execution for agent "${agent.name}" (ID: ${agent.id})`,
|
`Heartbeat execution for agent "${agent.name}" (ID: ${agent.id})`,
|
||||||
`Source: ${source}${triggerDetail ? ` (${triggerDetail})` : ""}`,
|
`Source: ${source}${triggerDetail ? ` (${triggerDetail})` : ""}`,
|
||||||
@@ -840,6 +947,7 @@ export class HeartbeatMonitor {
|
|||||||
taskDetail.description,
|
taskDetail.description,
|
||||||
"",
|
"",
|
||||||
taskDetail.prompt ? `PROMPT.md:\n${taskDetail.prompt}` : "No PROMPT.md available.",
|
taskDetail.prompt ? `PROMPT.md:\n${taskDetail.prompt}` : "No PROMPT.md available.",
|
||||||
|
...triggeringCommentLines,
|
||||||
"",
|
"",
|
||||||
"Review the task status and take appropriate action. Call heartbeat_done when finished.",
|
"Review the task status and take appropriate action. Call heartbeat_done when finished.",
|
||||||
].join("\n");
|
].join("\n");
|
||||||
@@ -1087,6 +1195,10 @@ export interface WakeContext {
|
|||||||
wakeReason: string;
|
wakeReason: string;
|
||||||
/** Detail about the specific trigger */
|
/** Detail about the specific trigger */
|
||||||
triggerDetail: string;
|
triggerDetail: string;
|
||||||
|
/** IDs of comments that triggered this wake (if any) */
|
||||||
|
triggeringCommentIds?: string[];
|
||||||
|
/** Type of comment that triggered this wake */
|
||||||
|
triggeringCommentType?: "steering" | "task" | "pr";
|
||||||
/** Budget governance status for the agent at trigger time */
|
/** Budget governance status for the agent at trigger time */
|
||||||
budgetStatus?: AgentBudgetStatus;
|
budgetStatus?: AgentBudgetStatus;
|
||||||
/** Additional context (intervalMs, etc.) */
|
/** Additional context (intervalMs, etc.) */
|
||||||
@@ -1130,13 +1242,15 @@ interface AgentTimer {
|
|||||||
export class HeartbeatTriggerScheduler {
|
export class HeartbeatTriggerScheduler {
|
||||||
private store: AgentStore;
|
private store: AgentStore;
|
||||||
private callback: TriggerCallback;
|
private callback: TriggerCallback;
|
||||||
|
private taskStore?: TaskStore;
|
||||||
private timers: Map<string, AgentTimer> = new Map();
|
private timers: Map<string, AgentTimer> = new Map();
|
||||||
private running = false;
|
private running = false;
|
||||||
private assignedListener: ((agent: import("@fusion/core").Agent, taskId: string) => void) | null = null;
|
private assignedListener: ((agent: import("@fusion/core").Agent, taskId: string) => void) | null = null;
|
||||||
|
|
||||||
constructor(store: AgentStore, callback: TriggerCallback) {
|
constructor(store: AgentStore, callback: TriggerCallback, taskStore?: TaskStore) {
|
||||||
this.store = store;
|
this.store = store;
|
||||||
this.callback = callback;
|
this.callback = callback;
|
||||||
|
this.taskStore = taskStore;
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
@@ -1261,11 +1375,39 @@ export class HeartbeatTriggerScheduler {
|
|||||||
// If getBudgetStatus fails, proceed without budget check
|
// If getBudgetStatus fails, proceed without budget check
|
||||||
}
|
}
|
||||||
|
|
||||||
|
let triggeringCommentIds: string[] | undefined;
|
||||||
|
if (this.taskStore && typeof this.taskStore.getTask === "function") {
|
||||||
|
try {
|
||||||
|
const [task, recentRuns] = await Promise.all([
|
||||||
|
this.taskStore.getTask(taskId),
|
||||||
|
this.store.getRecentRuns(agent.id, 1),
|
||||||
|
]);
|
||||||
|
|
||||||
|
const lastRunAt = recentRuns[0]?.startedAt;
|
||||||
|
const newSteeringComments = (task.steeringComments ?? []).filter((comment) =>
|
||||||
|
!lastRunAt || comment.createdAt > lastRunAt,
|
||||||
|
);
|
||||||
|
if (newSteeringComments.length > 0) {
|
||||||
|
triggeringCommentIds = newSteeringComments.map((comment) => comment.id);
|
||||||
|
}
|
||||||
|
} catch (error) {
|
||||||
|
heartbeatLog.warn(
|
||||||
|
`Failed to resolve triggering steering comments for assignment wake (${agent.id}/${taskId}): ${error instanceof Error ? error.message : String(error)}`,
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
heartbeatLog.log(`Assignment trigger for ${agent.id} (task: ${taskId})`);
|
heartbeatLog.log(`Assignment trigger for ${agent.id} (task: ${taskId})`);
|
||||||
await this.callback(agent.id, "assignment", {
|
await this.callback(agent.id, "assignment", {
|
||||||
taskId,
|
taskId,
|
||||||
wakeReason: "assignment",
|
wakeReason: "assignment",
|
||||||
triggerDetail: "task-assigned",
|
triggerDetail: "task-assigned",
|
||||||
|
...(triggeringCommentIds?.length
|
||||||
|
? {
|
||||||
|
triggeringCommentIds,
|
||||||
|
triggeringCommentType: "steering" as const,
|
||||||
|
}
|
||||||
|
: {}),
|
||||||
...(budgetStatus && { budgetStatus }),
|
...(budgetStatus && { budgetStatus }),
|
||||||
});
|
});
|
||||||
} catch (err) {
|
} catch (err) {
|
||||||
|
|||||||
@@ -266,9 +266,19 @@ export class InProcessRuntime
|
|||||||
source,
|
source,
|
||||||
triggerDetail: context.triggerDetail,
|
triggerDetail: context.triggerDetail,
|
||||||
taskId: typeof context.taskId === "string" ? context.taskId : undefined,
|
taskId: typeof context.taskId === "string" ? context.taskId : undefined,
|
||||||
|
triggeringCommentIds: Array.isArray(context.triggeringCommentIds)
|
||||||
|
? context.triggeringCommentIds.filter((id): id is string => typeof id === "string" && id.length > 0)
|
||||||
|
: undefined,
|
||||||
|
triggeringCommentType:
|
||||||
|
context.triggeringCommentType === "steering"
|
||||||
|
|| context.triggeringCommentType === "task"
|
||||||
|
|| context.triggeringCommentType === "pr"
|
||||||
|
? context.triggeringCommentType
|
||||||
|
: undefined,
|
||||||
contextSnapshot: { ...context },
|
contextSnapshot: { ...context },
|
||||||
});
|
});
|
||||||
},
|
},
|
||||||
|
this.taskStore,
|
||||||
);
|
);
|
||||||
this.triggerScheduler.start();
|
this.triggerScheduler.start();
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user