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

fix(FN-7344): hide stall badges during active agent logs

Browse Source
This commit is contained in:
gsxdsm
2026-06-30 23:34:43 -07:00
parent 4ab4aae02e
commit 4a1a04303f
6 changed files with 172 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
.changeset/fresh-agent-log-stall-badges.md Normal file
View File

@@ -0,0 +1,7 @@
---
"@runfusion/fusion": patch
---
summary: Stop showing stale in-review stall badges while agents are actively streaming logs.
category: fix
dev: TaskStore stall hydration now treats fresh buffered or persisted agent-log activity as active ownership.

13
packages/core/src/__tests__/store-in-review-stall.test.ts
View File

@@ -52,6 +52,19 @@ describe("TaskStore inReviewStall hydration", () => {
expect(task?.inReviewStall?.reason).toContain("no active merger");
});
it("omits merge-stalled hydration while fresh agent-log activity is streaming", async () => {
await seedTask("FN-7344", {});
await store.appendAgentLog("FN-7344", "rerunning merge verification", "thinking", undefined, "merger");
const listed = (await store.listTasks({ slim: true })).find((entry) => entry.id === "FN-7344");
expect(listed?.inReviewStall).toBeUndefined();
expect(listed?.inReviewStalled).toBeUndefined();
const detailed = await store.getTask("FN-7344");
expect(detailed.inReviewStall).toBeUndefined();
expect(detailed.inReviewStalled).toBeUndefined();
});
it("omits merge-stalled hydration while the task is already queued for merge", async () => {
await seedTask("FN-6088", {});
await store.enqueueMergeQueue("FN-6088");

17
packages/core/src/__tests__/store-stalled-review.test.ts
View File

@@ -57,6 +57,23 @@ describe("TaskStore stalledReview hydration", () => {
expect(detail.stalledReview?.matchCount).toBe(3);
});
it("omits stalledReview while fresh agent-log activity is streaming", async () => {
const task = await seedStalledInReviewTask();
const oldUpdatedAt = new Date(Date.now() - 6 * 60_000).toISOString();
const db = (store as unknown as { db: { prepare: (sql: string) => { run: (...params: unknown[]) => unknown } } }).db;
db.prepare("UPDATE tasks SET updatedAt = ? WHERE id = ?").run(oldUpdatedAt, task.id);
await store.appendAgentLog(task.id, "reviewer is comparing the squash against the branch", "thinking", undefined, "merger");
const slimTasks = await store.listTasks({ slim: true, column: "in-review" });
expect(slimTasks.find((entry) => entry.id === task.id)?.stalledReview).toBeUndefined();
const fullTasks = await store.listTasks({ slim: false, column: "in-review" });
expect(fullTasks.find((entry) => entry.id === task.id)?.stalledReview).toBeUndefined();
const detail = await store.getTask(task.id);
expect(detail.stalledReview).toBeUndefined();
});
it("omits stalledReview for tasks already queued for merge", async () => {
const task = await seedStalledInReviewTask();
await store.enqueueMergeQueue(task.id);

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

@@ -2,7 +2,7 @@ import { EventEmitter } from "node:events";
import { randomUUID } from "node:crypto";
import { mkdir, readdir, readFile, stat, writeFile, rename, unlink, rm } from "node:fs/promises";
import { join } from "node:path";
import { existsSync, watch, type Dirent, type FSWatcher } from "node:fs";
import { existsSync, statSync, watch, type Dirent, type FSWatcher } from "node:fs";
import { detectWorkspaceRepos, saveWorkspaceConfig, loadWorkspaceConfig } from "./git-repository.js";
import type { Task, TaskDetail, TaskCreateInput, TaskAttachment, AgentLogEntry, BoardConfig, Column, ColumnId, CheckoutClaimPrecondition, MergeResult, Settings, GlobalSettings, ProjectSettings, ActivityLogEntry, ActivityEventType, TaskDocument, TaskDocumentRevision, TaskDocumentCreateInput, TaskDocumentWithTask, Artifact, ArtifactCreateInput, ArtifactType, ArtifactWithTask, InboxTask, TaskLogEntry, RunMutationContext, RunAuditEvent, RunAuditEventInput, RunAuditEventFilter, ArchivedTaskEntry, ArchiveAgentLogMode, TaskPriority, SourceType, WorkflowStepTemplate, Agent, AutostashOrphanRecord, TaskCommitAssociation, TaskCommitAssociationMatchSource, TaskCommitAssociationConfidence, CommitAssociationDiffBackfillReport, GithubIssueAction, MergeQueueEntry, MergeQueueEnqueueOptions, MergeQueueAcquireOptions, MergeQueueReleaseOutcome, HandoffToReviewOptions, GoalCitation, GoalCitationFilter, GoalCitationInput, GoalCitationSurface, BranchGroup, BranchGroupCreateInput, BranchGroupUpdate, TaskBranchAssignmentMode, MergeRequestRecord, MergeRequestState, MergeRequestWorkflowProjectionOptions, CompletionHandoffMarker, WorkflowWorkItem, WorkflowWorkItemDueFilter, WorkflowWorkItemKind, WorkflowWorkItemState, WorkflowWorkItemTransitionPatch, WorkflowWorkItemUpsertInput, PrEntity, PrEntityCreateInput, PrEntityUpdate, PrEntityState, PrThreadState, PrThreadOutcome, PrConflictState, PrChecksRollup, PrReviewDecision, PluginActivation, PluginActivationInput } from "./types.js";
import { createActivityLogSnapshot, createRunAuditSnapshot, createTaskMetadataSnapshot, toTaskMetadataRecord, validateSnapshotEnvelope, type ActivityLogSnapshot, type RunAuditSnapshot, type TaskMetadataSnapshot } from "./shared-mesh-state.js";
@@ -163,7 +163,7 @@ import { SecretsStore } from "./secrets-store.js";
import { MasterKeyManager } from "./master-key.js";
import { hasSyncPassphraseConfigured } from "./secrets-sync-passphrase.js";
import { getTaskDoneBypassBlocker, getTaskMergeBlocker, resolveTaskMergeTarget } from "./task-merge.js";
import { getInReviewStallReason } from "./in-review-stall.js";
import { DEFAULT_STALE_MERGING_MIN_AGE_MS, getInReviewStallReason } from "./in-review-stall.js";
import { getInReviewStalledSignal } from "./in-review-stalled.js";
import { getStalePausedReviewSignal } from "./stale-paused-review.js";
import { getStalePausedTodoSignal } from "./stale-paused-todo.js";
@@ -174,6 +174,7 @@ import { createLogger } from "./logger.js";
import {
appendAgentLogEntriesSync,
countAgentLogEntries,
getAgentLogFilePath,
pruneAgentLogFiles as pruneAgentLogFileEntries,
readAgentLogEntries,
readAgentLogEntriesByTimeRange,
@@ -2688,6 +2689,53 @@ export class TaskStore extends EventEmitter<TaskStoreEvents> {
return total;
}
private getLatestAgentLogActivityMs(taskId: string): number | undefined {
let latest = Number.NEGATIVE_INFINITY;
for (let index = this.agentLogBuffer.length - 1; index >= 0; index -= 1) {
const entry = this.agentLogBuffer[index];
if (entry?.taskId !== taskId) continue;
const parsed = Date.parse(entry.timestamp);
if (Number.isFinite(parsed)) {
latest = Math.max(latest, parsed);
break;
}
}
try {
const filePath = getAgentLogFilePath(this.taskDir(taskId));
if (existsSync(filePath)) {
const fileMtimeMs = statSync(filePath).mtimeMs;
if (Number.isFinite(fileMtimeMs)) {
latest = Math.max(latest, fileMtimeMs);
}
}
} catch (error) {
storeLog.warn("Skipping agent-log freshness check for stalled badge hydration", {
taskId,
error: error instanceof Error ? error.message : String(error),
});
}
return Number.isFinite(latest) ? latest : undefined;
}
private hasFreshAgentLogActivitySinceTaskUpdate(task: Pick<Task, "id" | "column" | "updatedAt">, now: number): boolean {
if (task.column !== "in-review") return false;
const latestAgentLogMs = this.getLatestAgentLogActivityMs(task.id);
if (latestAgentLogMs == null) return false;
const updatedAtMs = Date.parse(task.updatedAt);
if (Number.isFinite(updatedAtMs) && latestAgentLogMs <= updatedAtMs) {
return false;
}
/*
FNXC:WorkflowLifecycle 2026-07-01-23:27:
In-review merge/review agents stream progress to agent-log JSONL without necessarily mutating the task row. Treat fresh agent-log writes as active ownership for stall-badge hydration so the board does not show Stalled/Merge stalled while a merger is visibly making progress.
*/
return Math.max(0, now - latestAgentLogMs) < DEFAULT_STALE_MERGING_MIN_AGE_MS;
}
private getTaskSelectClauseWithActivityLogLimit(limit: number): string {
const columns = [
"id", "lineageId", "title", "description", "priority", "\"column\"", "status", "size", "reviewLevel", "currentStep",
@@ -5352,11 +5400,14 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
const now = Date.now();
const settings = await this.getSettingsFast();
const mergeQueuedTaskIds = this.getMergeQueuedTaskIds();
const hasFreshAgentLogActivity = this.hasFreshAgentLogActivitySinceTaskUpdate(task, now);
const executingTaskIds = hasFreshAgentLogActivity ? new Set([task.id]) : undefined;
task.inReviewStall = mergeQueuedTaskIds.has(task.id)
? undefined
: getInReviewStallReason(task, {
now,
autoMerge: allowsAutoMergeProcessing(task, settings),
executingTaskIds,
engineActiveSinceMs: settings.engineActiveSinceMs,
engineActivationGraceMs: settings.engineActivationGraceMs,
});
@@ -5366,10 +5417,11 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
now,
thresholdMs: settings.inReviewStalledThresholdMs,
autoMerge: allowsAutoMergeProcessing(task, settings),
executingTaskIds,
engineActiveSinceMs: settings.engineActiveSinceMs,
engineActivationGraceMs: settings.engineActivationGraceMs,
});
task.stalledReview = mergeQueuedTaskIds.has(task.id) ? undefined : detectStalledReview(task, { now });
task.stalledReview = mergeQueuedTaskIds.has(task.id) || hasFreshAgentLogActivity ? undefined : detectStalledReview(task, { now });
// Derived at read time only; retrySummary is never persisted to SQLite.
task.retrySummary = computeRetrySummary(task);
@@ -5904,9 +5956,12 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
const activeTasks = await Promise.all((rows as unknown as TaskRow[]).map(async (row) => {
const task = this.rowToTask(row);
const isMergeQueued = mergeQueuedTaskIds.has(task.id);
const hasFreshAgentLogActivity = this.hasFreshAgentLogActivitySinceTaskUpdate(task, now);
const executingTaskIds = hasFreshAgentLogActivity ? new Set([task.id]) : undefined;
task.inReviewStall = isMergeQueued ? undefined : getInReviewStallReason(task, {
now,
autoMerge: allowsAutoMergeProcessing(task, settings),
executingTaskIds,
engineActiveSinceMs: settings.engineActiveSinceMs,
engineActivationGraceMs: settings.engineActivationGraceMs,
});
@@ -5920,6 +5975,7 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
now,
thresholdMs: settings.inReviewStalledThresholdMs,
autoMerge: allowsAutoMergeProcessing(task, settings),
executingTaskIds,
engineActiveSinceMs: settings.engineActiveSinceMs,
engineActivationGraceMs: settings.engineActivationGraceMs,
});
@@ -5948,7 +6004,7 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
}
}
}
task.stalledReview = isMergeQueued ? undefined : detectStalledReview(task, { now });
task.stalledReview = isMergeQueued || hasFreshAgentLogActivity ? undefined : detectStalledReview(task, { now });
// Derived at read time only; retrySummary is never persisted to SQLite.
task.retrySummary = computeRetrySummary(task);
@@ -6423,9 +6479,12 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
const tasks = rows.slice(0, resolvedLimit).map((row) => {
const task = this.rowToTask(row);
const isMergeQueued = mergeQueuedTaskIds.has(task.id);
const hasFreshAgentLogActivity = this.hasFreshAgentLogActivitySinceTaskUpdate(task, now);
const executingTaskIds = hasFreshAgentLogActivity ? new Set([task.id]) : undefined;
task.inReviewStall = isMergeQueued ? undefined : getInReviewStallReason(task, {
now,
autoMerge: allowsAutoMergeProcessing(task, settings),
executingTaskIds,
engineActiveSinceMs: settings.engineActiveSinceMs,
engineActivationGraceMs: settings.engineActivationGraceMs,
});
@@ -6439,6 +6498,7 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
now,
thresholdMs: settings.inReviewStalledThresholdMs,
autoMerge: allowsAutoMergeProcessing(task, settings),
executingTaskIds,
engineActiveSinceMs: settings.engineActiveSinceMs,
engineActivationGraceMs: settings.engineActivationGraceMs,
});
@@ -6468,7 +6528,7 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
}
}
task.timedExecutionMs = this.computeTimedExecutionMs(task.log);
task.stalledReview = isMergeQueued ? undefined : detectStalledReview(task, { now });
task.stalledReview = isMergeQueued || hasFreshAgentLogActivity ? undefined : detectStalledReview(task, { now });
// Derived at read time only; retrySummary is never persisted to SQLite.
task.retrySummary = computeRetrySummary(task);
task.log = [];
@@ -6588,9 +6648,12 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
const activeMatches = await Promise.all(rows.map(async (row) => {
const task = this.rowToTask(row);
const isMergeQueued = mergeQueuedTaskIds.has(task.id);
const hasFreshAgentLogActivity = this.hasFreshAgentLogActivitySinceTaskUpdate(task, now);
const executingTaskIds = hasFreshAgentLogActivity ? new Set([task.id]) : undefined;
task.inReviewStall = isMergeQueued ? undefined : getInReviewStallReason(task, {
now,
autoMerge: allowsAutoMergeProcessing(task, settings),
executingTaskIds,
engineActiveSinceMs: settings.engineActiveSinceMs,
engineActivationGraceMs: settings.engineActivationGraceMs,
});
@@ -6604,6 +6667,7 @@ ${TASK_UPSERT_SQL_ASSIGNMENTS}
now,
thresholdMs: settings.inReviewStalledThresholdMs,
autoMerge: allowsAutoMergeProcessing(task, settings),
executingTaskIds,
engineActiveSinceMs: settings.engineActiveSinceMs,
engineActivationGraceMs: settings.engineActivationGraceMs,
});

51
packages/dashboard/app/hooks/useTasks.ts
View File

@@ -1,5 +1,5 @@
import { useState, useEffect, useCallback, useRef } from "react";
import type { Task, Column, ColumnId, TaskCreateInput, MergeResult, GithubIssueAction } from "@fusion/core";
import type { Task, Column, ColumnId, TaskCreateInput, MergeResult, GithubIssueAction, AgentLogEntry } from "@fusion/core";
import { normalizeColumn } from "@fusion/core";
import * as api from "../api";
import { subscribeSse } from "../sse-bus";
@@ -30,6 +30,33 @@ function filterActiveTasks(tasks: Task[]): Task[] {
return tasks.filter((task) => !isSoftDeleted(task));
}
type AgentLogActivityEvent = Pick<AgentLogEntry, "taskId" | "timestamp" | "type" | "agent">;
function clearInReviewStallForFreshAgentLog(task: Task, entry: AgentLogActivityEvent): Task {
if (task.id !== entry.taskId || task.column !== "in-review") return task;
const logTimestampMs = Date.parse(entry.timestamp);
const taskUpdatedAtMs = Date.parse(task.updatedAt);
if (
Number.isFinite(logTimestampMs) &&
Number.isFinite(taskUpdatedAtMs) &&
logTimestampMs <= taskUpdatedAtMs
) {
return task;
}
if (!task.inReviewStall && !task.inReviewStalled && !task.stalledReview) return task;
/*
FNXC:DashboardStallBadges 2026-07-01-23:44:
Board cards must not show Stalled/Merge stalled while an in-review agent is actively writing logs. The task row can remain unchanged during merger/reviewer work, so fresh agent-log metadata clears only derived stall badge fields until the next authoritative task refresh.
*/
return {
...task,
inReviewStall: undefined,
inReviewStalled: undefined,
stalledReview: undefined,
};
}
/**
* Compare two ISO timestamp strings.
* Returns positive if a is newer than b, negative if b is newer, 0 if equal.
@@ -495,6 +522,27 @@ export function useTasks(options?: UseTasksOptions) {
});
};
const handleAgentLog = (e: MessageEvent) => {
if (isStale()) {
traceDroppedStaleEvent();
return;
}
if (searchQueryRef.current) {
return;
}
const entry = JSON.parse(e.data) as AgentLogActivityEvent;
if (!entry.taskId || !entry.timestamp) return;
setTasks((prev) => {
let changed = false;
const next = prev.map((task) => {
const cleared = clearInReviewStallForFreshAgentLog(task, entry);
if (cleared !== task) changed = true;
return cleared;
});
return changed ? next : prev;
});
};
const unsubscribe = subscribeSse(`/api/events${query}`, {
events: {
"task:created": handleCreated,
@@ -502,6 +550,7 @@ export function useTasks(options?: UseTasksOptions) {
"task:updated": handleUpdated,
"task:deleted": handleDeleted,
"task:merged": handleMerged,
"agent:log": handleAgentLog,
},
// Guard onReconnect against stale SSE callbacks: do not call refreshTasks
// if the SSE was disabled or the effect unmounted while reconnect was pending.

16
packages/dashboard/src/sse.ts
View File

@@ -11,6 +11,7 @@ import type {
FixFeatureCreatedPayload,
ChatStore,
AutomationStore,
AgentLogEntry,
} from "@fusion/core";
import type { AiSessionStore } from "./ai-session-store.js";
@@ -530,6 +531,15 @@ export function createSSE(
const onMerged = (result: unknown) => {
send(`event: task:merged\ndata: ${JSON.stringify(stripTaskEventHeavyFields(result))}\n\n`);
};
const onAgentLog = (entry: AgentLogEntry) => {
const payload = {
taskId: entry.taskId,
timestamp: entry.timestamp,
type: entry.type,
agent: entry.agent,
};
send(`event: agent:log\ndata: ${JSON.stringify(payload)}\n\n`);
};
const onArtifactRegistered = (artifact: unknown) => {
/* FNXC:ArtifactRegistry 2026-06-27-00:00: Forward TaskStore's authoritative artifact registration event so live artifact surfaces refresh even when the best-effort inbox notification is absent or delayed. */
@@ -827,6 +837,7 @@ export function createSSE(
store.off("task:updated", onUpdated);
store.off("task:deleted", onDeleted);
store.off("task:merged", onMerged);
store.off("agent:log", onAgentLog);
store.off("artifact:registered", onArtifactRegistered);
if (missionStore) {
missionStore.off("mission:created", onMissionCreated);
@@ -937,6 +948,11 @@ export function createSSE(
store.on("task:updated", onUpdated);
store.on("task:deleted", onDeleted);
store.on("task:merged", onMerged);
/*
FNXC:DashboardStallBadges 2026-07-01-23:42:
Agent log streaming is authoritative evidence that an in-review agent is active even when the task row has not changed. Forward compact log metadata on the board stream so clients can clear false Stalled/Merge stalled badges without rewriting the full task for every log line.
*/
store.on("agent:log", onAgentLog);
store.on("artifact:registered", onArtifactRegistered);
if (missionStore) {