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

FN-6177: remove queued-concurrency scheduler audit

Browse Source 
Remove the write-only queued-concurrency scheduler audit and its stale documentation.

- delete scheduler:dispatch-queued-concurrency audit emission and memo bookkeeping
- keep queued concurrency log dedupe coverage while removing audit-specific scheduler test assertions
- update architecture docs to describe queued concurrency diagnostics as log output only

Files changed:
 docs/architecture.md                            |  4 +-
 packages/engine/src/__tests__/scheduler.test.ts | 33 ----------------
 packages/engine/src/scheduler.ts                | 51 -------------------------
 3 files changed, 2 insertions(+), 86 deletions(-)

Fusion-Task-Id: FN-6177

Fusion-Task-Lineage: e4516b41-5525-4e26-9602-772302d63f3d
This commit is contained in:
gsxdsm
2026-06-10 00:39:22 -07:00
parent 5e35385594
commit 5b5cd77021
3 changed files with 2 additions and 86 deletions
Download Patch File Download Diff File Expand all files Collapse all files

4
docs/architecture.md
View File

@@ -1060,7 +1060,7 @@ Mesh configuration and post-provision managed-node operations are registered sep
### Run Audit API
The run-audit system records every mutation performed by the engine across four domains:
- **Database** — task:create, task:update, task:move, etc. Node handoff/recovery emits structured events: `node:handoff:parked` (handoff denied/parked), `node:handoff:reassign-local` (local takeover approved), `node:handoff:reassign-any` (any-healthy takeover approved), and `node:lease:recovered` (abandoned lease cleared and task requeued). Scheduler dispatch contention also emits `scheduler:dispatch-queued-concurrency` with transition-only cadence: emit once when a task first hits a stable limiter signature (`bindingGates`), suppress while unchanged, clear the memo when the task leaves that queued-concurrency state, and emit again only after the limiter signature changes or reappears. Metadata includes `bindingGates` (`maxConcurrent`/`maxWorktrees`/`semaphore`), per-gate `{ used, limit, slack }`, `holders`, and computed `available`.
- **Database** — task:create, task:update, task:move, etc. Node handoff/recovery emits structured events: `node:handoff:parked` (handoff denied/parked), `node:handoff:reassign-local` (local takeover approved), `node:handoff:reassign-any` (any-healthy takeover approved), and `node:lease:recovered` (abandoned lease cleared and task requeued).
- **Git** — worktree:create, commit:create, merge:resolve, merge:audit-failure, `worktree:reanchored`, and worktrunk lifecycle events (`worktree:worktrunk-install|create|sync|prune|remove`, plus `worktree:worktrunk-fallback`, `worktree:worktrunk-failure`, and `worktree:worktrunk-fallback-native`). Worktrunk events share metadata `{ op, binaryPath?, worktreePath?, durationMs?, exitCode?, stderrPreview?, installSource?, prunedCount? }` with `installSource` (`"release-binary" | "cargo"`) limited to successful `worktree:worktrunk-install` events and `prunedCount` limited to successful prune events when known. `worktree:worktrunk-install` is emitted only for true install actions; cache hits, configured `worktrunk.binaryPath` overrides, and `$PATH` resolutions intentionally remain silent. Dirty post-merge audit outcomes emit `merge:audit-failure` with metadata `{ mode, strategy, action, reason, issueCount, duplicateSubjectCount, touchedFileOverlapCount, verificationPassed, auditTargetLabel }`. FN-5279 adds `merge:reuse-handoff-acquired`, `merge:reuse-handoff-refused`, `merge:reuse-handoff-released`, and `merge:reuse-handoff-deferred-to-worktrunk` for task-worktree auto-merge handoff visibility. FN-5351 adds `merge:integration-worktree-state` (pre-handoff checkout/dirty snapshot for resolved integration branch), `merge:cwd-integration-fallback-refused` (terminal refusal park event), and `merge:integration-ref-advance` (integration ref advance outcome telemetry).
- **Git / `merge:file-scope-violation`** — emitted by the merger when `FileScopeViolationError` aborts a squash. `target` is the task ID; metadata includes `stagedFiles`, `declaredScope`, `resetLabel`, `stagedFileCount`, and `declaredScopeCount`. Consumed by `fileScopeInvariantFailuresPerDay` in `GET /api/health/reliability` (FN-4360).
- **Git / `merge:no-op-attribution-mismatch`** — emitted by the rebase landed-files attribution guard (FN-5304) when `<rebaseBaseSha>..HEAD` has zero attributable own commits but the source `fusion/<id>` tip still carries attributable own commits. `target` is the task ID; metadata includes `recordedSha`, `rebaseMergeBaseSha`, `sourceBranchRef`, `sourceBranchOwnCommitCount`, and `sourceBranchOwnCommitShas`.
@@ -1075,7 +1075,7 @@ The run-audit system records every mutation performed by the engine across four
Events are tied to specific run IDs for end-to-end traceability.
For scheduler concurrency diagnostics, the queued reason now names the active limiter(s) and usage (for example `gate=maxConcurrent ...`). Read `metadata.bindingGates` first to identify the limiter. `holders.maxConcurrent` and `holders.maxWorktrees` are current `in-progress` task IDs; `holders.semaphore` mirrors that set but semaphore slots can also be consumed by triage/merge agents outside `in-progress`. So if `semaphore.used` exceeds the visible holder list, that usually indicates non-execution agents are legitimately consuming shared capacity (not stale accounting). These events are transition-only: a newly emitted line/event indicates the limiter signature changed or the condition cleared and later reappeared, not that a poll loop simply observed the same blocked state again.
For scheduler concurrency diagnostics, the queued reason now names the active limiter(s) and usage (for example `gate=maxConcurrent ...`). The reason includes the `bindingGates` (`maxConcurrent`/`maxWorktrees`/`semaphore`), per-gate `{ used, limit, slack }`, `holders`, and computed `available`. `holders.maxConcurrent` and `holders.maxWorktrees` are current `in-progress` task IDs; `holders.semaphore` mirrors that set but semaphore slots can also be consumed by triage/merge agents outside `in-progress`. So if `semaphore.used` exceeds the visible holder list, that usually indicates non-execution agents are legitimately consuming shared capacity (not stale accounting). These queued-reason logs are transition-only: a newly emitted line indicates the limiter signature changed or the condition cleared and later reappeared, not that a poll loop simply observed the same blocked state again.
**Run audit endpoints:**
- `GET /api/agents/:id/runs/:runId/audit` — Returns audit trail for a specific agent run

33
packages/engine/src/__tests__/scheduler.test.ts
View File

@@ -1834,12 +1834,6 @@ describe("Scheduler", () => {
(call: unknown[]) => call[0] === "FN-C" && String(call[1]).includes("queued — concurrency limit reached"),
);
expect(concurrencyReasonCalls).toHaveLength(1);
const auditCalls = (store.recordRunAuditEvent as ReturnType<typeof vi.fn>).mock.calls.filter(
(call: unknown[]) => (call[0] as { mutationType?: string } | undefined)?.mutationType === "scheduler:dispatch-queued-concurrency",
);
expect(auditCalls).toHaveLength(1);
expect(auditCalls[0]?.[0]?.metadata?.bindingGates).toEqual(["maxConcurrent"]);
});
it("dedupes queued-concurrency logs when only non-binding semaphore counts change", async () => {
@@ -1872,12 +1866,6 @@ describe("Scheduler", () => {
(call: unknown[]) => call[0] === "FN-D" && String(call[1]).includes("queued — concurrency limit reached"),
);
expect(concurrencyReasonCalls).toHaveLength(1);
const auditCalls = (store.recordRunAuditEvent as ReturnType<typeof vi.fn>).mock.calls.filter(
(call: unknown[]) => (call[0] as { mutationType?: string } | undefined)?.mutationType === "scheduler:dispatch-queued-concurrency",
);
expect(auditCalls).toHaveLength(1);
expect(auditCalls[0]?.[0]?.metadata?.bindingGates).toEqual(["maxWorktrees"]);
});
it("dedupes queued-concurrency logs across used/limit churn on the same binding gate", async () => {
@@ -1910,13 +1898,6 @@ describe("Scheduler", () => {
);
expect(concurrencyReasonCalls).toHaveLength(1);
expect(String(concurrencyReasonCalls[0]?.[1])).toContain("semaphore used=1/2");
const auditCalls = (store.recordRunAuditEvent as ReturnType<typeof vi.fn>).mock.calls.filter(
(call: unknown[]) => (call[0] as { mutationType?: string } | undefined)?.mutationType === "scheduler:dispatch-queued-concurrency",
);
expect(auditCalls).toHaveLength(1);
expect(auditCalls[0]?.[0]?.metadata?.bindingGates).toEqual(["semaphore"]);
expect(auditCalls[0]?.[0]?.metadata?.semaphore).toEqual({ used: 1, limit: 2, slack: 1 });
});
it("suppresses re-log and re-audit when only binding holder identity changes", async () => {
@@ -1950,13 +1931,6 @@ describe("Scheduler", () => {
(call: unknown[]) => call[0] === "FN-D" && String(call[1]).includes("queued — concurrency limit reached"),
);
expect(concurrencyReasonCalls).toHaveLength(1);
const auditCalls = (store.recordRunAuditEvent as ReturnType<typeof vi.fn>).mock.calls.filter(
(call: unknown[]) => (call[0] as { mutationType?: string } | undefined)?.mutationType === "scheduler:dispatch-queued-concurrency",
);
expect(auditCalls).toHaveLength(1);
expect(auditCalls[0]?.[0]?.metadata?.bindingGates).toEqual(["maxWorktrees"]);
expect(auditCalls[0]?.[0]?.metadata?.holders?.maxWorktrees).toEqual(["FN-A"]);
});
it("re-logs and re-audits when binding gate changes", async () => {
@@ -1995,13 +1969,6 @@ describe("Scheduler", () => {
expect(concurrencyReasonCalls).toHaveLength(2);
expect(String(concurrencyReasonCalls[0]?.[1])).toContain("gate=maxConcurrent");
expect(String(concurrencyReasonCalls[1]?.[1])).toContain("gate=maxWorktrees");
const auditCalls = (store.recordRunAuditEvent as ReturnType<typeof vi.fn>).mock.calls.filter(
(call: unknown[]) => (call[0] as { mutationType?: string } | undefined)?.mutationType === "scheduler:dispatch-queued-concurrency",
);
expect(auditCalls).toHaveLength(2);
expect(auditCalls[0]?.[0]?.metadata?.bindingGates).toEqual(["maxConcurrent"]);
expect(auditCalls[1]?.[0]?.metadata?.bindingGates).toEqual(["maxWorktrees"]);
});
it("formats queued-concurrency memo keys from binding gates only", () => {

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

@@ -502,8 +502,6 @@ export class Scheduler {
private wasPermanentAgentUnavailable = new Set<string>();
/** Tracks dispatch-queued reason signatures to avoid per-tick log spam. */
private wasDispatchQueuedReasonLogged = new Set<string>();
/** Tracks the last stable concurrency-block signature emitted for a task. */
private dispatchQueuedConcurrencyAuditMemo = new Map<string, string>();
/** Tracks per-task candidacy fingerprints for task:updated auto-claim invalidation gating. */
private lastAutoClaimFingerprint = new Map<string, string>();
/** Tracks recent engine-sourced in-progress → todo requeues to prevent immediate re-dispatch races. */
@@ -791,7 +789,6 @@ export class Scheduler {
this.wasNodeBlocked.delete(task.id);
this.wasPermanentAgentUnavailable.delete(task.id);
this.clearDispatchQueuedReasonMemo(task.id);
this.clearDispatchQueuedConcurrencyAuditMemo(task.id);
void (async () => {
try {
@@ -936,7 +933,6 @@ export class Scheduler {
this.wasNodeDispatchValidationBlocked.clear();
this.wasPermanentAgentUnavailable.clear();
this.wasDispatchQueuedReasonLogged.clear();
this.dispatchQueuedConcurrencyAuditMemo.clear();
schedulerLog.log("Stopped");
}
@@ -955,27 +951,11 @@ export class Scheduler {
}
this.clearDispatchQueuedReasonMemo(taskId);
if (!key.includes(":queued-concurrency:")) {
this.clearDispatchQueuedConcurrencyAuditMemo(taskId);
}
this.wasDispatchQueuedReasonLogged.add(key);
await this.store.logEntry(taskId, reason);
return true;
}
private shouldEmitDispatchQueuedConcurrencyAudit(taskId: string, signature: string): boolean {
const lastSignature = this.dispatchQueuedConcurrencyAuditMemo.get(taskId);
if (lastSignature === signature) {
return false;
}
this.dispatchQueuedConcurrencyAuditMemo.set(taskId, signature);
return true;
}
private clearDispatchQueuedConcurrencyAuditMemo(taskId: string): void {
this.dispatchQueuedConcurrencyAuditMemo.delete(taskId);
}
private emitDependencyParityDiff(diff: SchedulingDependencyParityDiff): void {
void this.store.recordRunAuditEvent?.({
taskId: diff.taskId,
@@ -992,33 +972,6 @@ export class Scheduler {
});
}
private async emitDispatchQueuedConcurrencyAudit(task: Task, diagnostic: ConcurrencyGateDiagnostic): Promise<void> {
try {
await this.store.recordRunAuditEvent?.({
taskId: task.id,
agentId: "scheduler",
runId: generateSyntheticRunId("scheduler", task.id),
domain: "database",
mutationType: "scheduler:dispatch-queued-concurrency",
target: task.id,
metadata: {
bindingGates: diagnostic.bindingGates,
maxConcurrent: diagnostic.maxConcurrentGate,
maxWorktrees: diagnostic.maxWorktreesGate,
semaphore: diagnostic.semaphoreGate,
holders: diagnostic.holders,
available: diagnostic.available,
// U6: additive per-column capacity gates (present only flag-ON).
...(diagnostic.perColumnGates ? { perColumnGates: diagnostic.perColumnGates } : {}),
},
});
} catch (error) {
schedulerLog.warn(
`Task ${task.id} failed to emit dispatch queued concurrency audit: ${error instanceof Error ? error.message : String(error)}`,
);
}
}
private async emitNodeUnreachableRecoveryAudit(
task: Task,
metadata: {
@@ -1676,9 +1629,6 @@ export class Scheduler {
reason,
concurrencySignature,
);
if (this.shouldEmitDispatchQueuedConcurrencyAudit(task.id, concurrencySignature)) {
await this.emitDispatchQueuedConcurrencyAudit(task, concurrencyGateDiagnostic);
}
continue;
}
@@ -2014,7 +1964,6 @@ export class Scheduler {
this.wasNodeDispatchValidationBlocked.delete(task.id);
this.wasPermanentAgentUnavailable.delete(task.id);
this.clearDispatchQueuedReasonMemo(task.id);
this.clearDispatchQueuedConcurrencyAuditMemo(task.id);
await this.store.logEntry(task.id, `Node routing resolved: ${effectiveNode.nodeId ?? "local"} (source: ${effectiveNode.source})`);
this.options.onSchedule?.(task);
started++;