feat(FN-1496): harden semaphore lane-vs-utility boundary
- Add explicit boundary comments to serve.ts and dashboard.ts clarifying semaphore lane usage - Add regression tests for semaphore lane-vs-utility boundary in serve.test.ts and dashboard.test.ts - Add changeset for @gsxdsm/fusion patch release
This commit is contained in:
gsxdsm
@@ -102,12 +102,17 @@ export async function runServe(
|
||||
taskStore: store,
|
||||
});
|
||||
|
||||
// ── HeartbeatMonitor: runtime monitoring and execution for agents ───
|
||||
// ── HeartbeatMonitor: runtime monitoring (UTILITY — NO semaphore) ───
|
||||
//
|
||||
// ⚠️ UTILITY PATH: This component does NOT receive the task-lane semaphore.
|
||||
//
|
||||
// Provides the Paperclip-style heartbeat execution engine:
|
||||
// wake → check inbox → work → exit
|
||||
//
|
||||
// Enables agent execution runs triggered by timers, assignments, or manual API calls.
|
||||
// Enables lightweight agent sessions for monitoring, not task-lane work.
|
||||
// By design, heartbeat sessions are independent of task concurrency limits
|
||||
// so they can run regardless of how busy the task lanes are.
|
||||
//
|
||||
// Passed to createServer to enable the heartbeat routes.
|
||||
//
|
||||
let heartbeatMonitor: HeartbeatMonitor | undefined;
|
||||
@@ -127,7 +132,13 @@ export async function runServe(
|
||||
});
|
||||
heartbeatMonitor.start();
|
||||
|
||||
// HeartbeatTriggerScheduler manages timer and assignment-based triggers
|
||||
// HeartbeatTriggerScheduler: trigger scheduling (UTILITY — NO semaphore) ──
|
||||
//
|
||||
// ⚠️ UTILITY PATH: This scheduler does NOT receive the task-lane semaphore.
|
||||
//
|
||||
// Manages timer and assignment-based triggers for heartbeat execution.
|
||||
// By design, trigger scheduling is independent of task-lane concurrency limits.
|
||||
//
|
||||
triggerScheduler = new HeartbeatTriggerScheduler(
|
||||
agentStore,
|
||||
async (agentId, source, context: WakeContext) => {
|
||||
@@ -194,6 +205,29 @@ export async function runServe(
|
||||
console.log("[engine] Starting in paused mode — automation disabled");
|
||||
}
|
||||
|
||||
// ── Task-lane concurrency semaphore ────────────────────────────────
|
||||
//
|
||||
// ⚠️ SEMAPHORE BOUNDARY: This semaphore governs ONLY task-lane agents.
|
||||
//
|
||||
// Governed components (task lanes):
|
||||
// - TriageProcessor: specification agents that produce PROMPT.md
|
||||
// - TaskExecutor: task execution agents that implement features
|
||||
// - Scheduler: coordinates which agent gets which task
|
||||
// - onMerge: AI-powered merge execution for completed tasks
|
||||
//
|
||||
// UTILITY WORKFLOWS — NOT governed by this semaphore:
|
||||
// - HeartbeatMonitor: lightweight heartbeat sessions for agent monitoring
|
||||
// - HeartbeatTriggerScheduler: timer/assignment-based trigger scheduling
|
||||
// - CronRunner (via createAiPromptExecutor): scheduled automation prompts
|
||||
// - Model sync, auth setup, plugin loading: bootstrap/setup workflows
|
||||
//
|
||||
// This boundary prevents utility workflows from being blocked by
|
||||
// task-lane saturation and ensures utility work is always available.
|
||||
//
|
||||
// The limit is read from a cached value that is refreshed from the store
|
||||
// on each scheduler poll cycle (see engine block below). This avoids
|
||||
// async I/O in the synchronous getter while still picking up live changes.
|
||||
//
|
||||
const initialSettings = await store.getSettings();
|
||||
let cachedMaxConcurrent = initialSettings.maxConcurrent;
|
||||
const semaphore = new AgentSemaphore(() => cachedMaxConcurrent);
|
||||
@@ -216,6 +250,15 @@ export async function runServe(
|
||||
const usageLimitPauser = new UsageLimitPauser(store);
|
||||
const githubClient = new GitHubClient(process.env.GITHUB_TOKEN);
|
||||
|
||||
// ── onMerge: AI-powered merge (TASK LANE — semaphore-gated) ─────────────
|
||||
//
|
||||
// ⚠️ TASK LANE: aiMergeTask is wrapped with semaphore.run() to ensure
|
||||
// merge agents count toward settings.maxConcurrent alongside triage and execution.
|
||||
//
|
||||
// The raw aiMergeTask does NOT receive the semaphore directly;
|
||||
// the semaphore gating is applied at the onMerge wrapper level.
|
||||
//
|
||||
// Track the active merge session so it can be killed on global pause.
|
||||
let activeMergeSession: { dispose: () => void } | null = null;
|
||||
|
||||
const rawMerge = (taskId: string) =>
|
||||
@@ -547,6 +590,11 @@ export async function runServe(
|
||||
},
|
||||
});
|
||||
|
||||
// ── TriageProcessor: task specification (TASK LANE — receives semaphore) ──
|
||||
//
|
||||
// Receives the task-lane semaphore to ensure specification agents
|
||||
// count toward settings.maxConcurrent alongside execution and merge.
|
||||
//
|
||||
const triage = new TriageProcessor(store, cwd, {
|
||||
semaphore,
|
||||
usageLimitPauser,
|
||||
@@ -558,6 +606,11 @@ export async function runServe(
|
||||
});
|
||||
triageRef.current = triage;
|
||||
|
||||
// ── TaskExecutor: task execution (TASK LANE — receives semaphore) ──────────
|
||||
//
|
||||
// Receives the task-lane semaphore to ensure execution agents
|
||||
// count toward settings.maxConcurrent alongside specification and merge.
|
||||
//
|
||||
const executor = new TaskExecutor(store, cwd, {
|
||||
semaphore,
|
||||
pool,
|
||||
@@ -577,6 +630,11 @@ export async function runServe(
|
||||
prCommentHandler.handleNewComments(taskId, prInfo, comments),
|
||||
);
|
||||
|
||||
// ── Scheduler: task coordination (TASK LANE — receives semaphore) ──────────
|
||||
//
|
||||
// Receives the task-lane semaphore to ensure task assignment decisions
|
||||
// respect the concurrency limit alongside running execution agents.
|
||||
//
|
||||
const scheduler = new Scheduler(store, {
|
||||
semaphore,
|
||||
prMonitor,
|
||||
@@ -639,6 +697,17 @@ export async function runServe(
|
||||
}
|
||||
};
|
||||
|
||||
// ── CronRunner: scheduled automation (UTILITY — NO semaphore) ──────────
|
||||
//
|
||||
// ⚠️ UTILITY PATH: CronRunner does NOT receive the task-lane semaphore.
|
||||
//
|
||||
// Uses createAiPromptExecutor (cwd-only factory) for AI execution in
|
||||
// scheduled tasks. By design, automation prompts are independent of
|
||||
// task concurrency limits so they can run regardless of task-lane saturation.
|
||||
//
|
||||
// createAiPromptExecutor takes only `cwd` (no semaphore parameter),
|
||||
// ensuring automation never competes with task-lane agents for slots.
|
||||
//
|
||||
const aiPromptExecutor = await createAiPromptExecutor(cwd);
|
||||
const cronRunner = new CronRunner(store, automationStore, {
|
||||
aiPromptExecutor,
|
||||
|
||||
Reference in New Issue
Block a user