diff --git a/.changeset/fresh-agent-log-stall-badges.md b/.changeset/fresh-agent-log-stall-badges.md new file mode 100644 index 0000000000..3ba30eaffe --- /dev/null +++ b/.changeset/fresh-agent-log-stall-badges.md @@ -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. diff --git a/packages/core/src/__tests__/store-in-review-stall.test.ts b/packages/core/src/__tests__/store-in-review-stall.test.ts index a417c5eac2..a48e7d6927 100644 --- a/packages/core/src/__tests__/store-in-review-stall.test.ts +++ b/packages/core/src/__tests__/store-in-review-stall.test.ts @@ -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"); diff --git a/packages/core/src/__tests__/store-stalled-review.test.ts b/packages/core/src/__tests__/store-stalled-review.test.ts index 24e5eacd5d..1533f7c5a6 100644 --- a/packages/core/src/__tests__/store-stalled-review.test.ts +++ b/packages/core/src/__tests__/store-stalled-review.test.ts @@ -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); diff --git a/packages/core/src/store.ts b/packages/core/src/store.ts index 30a09d595a..b143d6da99 100644 --- a/packages/core/src/store.ts +++ b/packages/core/src/store.ts @@ -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 { 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, 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, }); diff --git a/packages/dashboard/app/hooks/useTasks.ts b/packages/dashboard/app/hooks/useTasks.ts index dcbc405dad..7752959ff1 100644 --- a/packages/dashboard/app/hooks/useTasks.ts +++ b/packages/dashboard/app/hooks/useTasks.ts @@ -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; + +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. diff --git a/packages/dashboard/src/sse.ts b/packages/dashboard/src/sse.ts index f02b9f5775..14d76f194e 100644 --- a/packages/dashboard/src/sse.ts +++ b/packages/dashboard/src/sse.ts @@ -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) {