diff --git a/docs/architecture.md b/docs/architecture.md index 20fd9603cb..0bba41d2e2 100644 --- a/docs/architecture.md +++ b/docs/architecture.md @@ -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 `..HEAD` has zero attributable own commits but the source `fusion/` 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 diff --git a/packages/engine/src/__tests__/scheduler.test.ts b/packages/engine/src/__tests__/scheduler.test.ts index b670577900..64dafc8180 100644 --- a/packages/engine/src/__tests__/scheduler.test.ts +++ b/packages/engine/src/__tests__/scheduler.test.ts @@ -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).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).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).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).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).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", () => { diff --git a/packages/engine/src/scheduler.ts b/packages/engine/src/scheduler.ts index bca97e97ee..c945550542 100644 --- a/packages/engine/src/scheduler.ts +++ b/packages/engine/src/scheduler.ts @@ -502,8 +502,6 @@ export class Scheduler { private wasPermanentAgentUnavailable = new Set(); /** Tracks dispatch-queued reason signatures to avoid per-tick log spam. */ private wasDispatchQueuedReasonLogged = new Set(); - /** Tracks the last stable concurrency-block signature emitted for a task. */ - private dispatchQueuedConcurrencyAuditMemo = new Map(); /** Tracks per-task candidacy fingerprints for task:updated auto-claim invalidation gating. */ private lastAutoClaimFingerprint = new Map(); /** 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 { - 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++;