diff --git a/DESIGN.md b/DESIGN.md index 16f428bffa7..d80f6a70754 100644 --- a/DESIGN.md +++ b/DESIGN.md @@ -48,7 +48,10 @@ alignment in every step and conditional state, not just the first screen. Do not show a toast for task or run state already visible on the current screen. This includes descendant runs represented by the open subtree. Show local action results in place; keep failures actionable inline. Notifications for other work -remain useful. Expected cancellation is neutral gray, not an error. The composer's Stop action stops the current response and leaves the composer available for a new message. Pause work is a separate explicit task or subtree action. A paused task replaces the composer with an amber takeover. It says “Task is +remain useful. Repeated delivery of the same run outcome must refresh cached state +without repeating its toast, including after reconnecting. A terminal outcome +delivered more than five minutes after the run finished is historical and should +refresh state silently. Expected cancellation is neutral gray, not an error. The composer's Stop action stops the current response and leaves the composer available for a new message. Pause work is a separate explicit task or subtree action. A paused task replaces the composer with an amber takeover. It says “Task is paused.” and “Resume this task to send a message.” with a “Resume task” action. Subtrees use “Subtree is paused.” and “Resume subtree.” The takeover cannot be dismissed, retains drafts, and hides message inputs until the pause is released. diff --git a/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts b/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts index 4eb79127f5e..aeaa9f0f352 100644 --- a/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts +++ b/server/src/services/native-runtime/native-run-finalizer-telemetry.test.ts @@ -48,6 +48,7 @@ function captureRunFailureCallsFrom(fromIndex: number) { import { PaperclipControlPlanePort } from "./paperclip-control-plane-port.js"; import { finalizeNativeRun, recordNativeFinalizationFailure } from "./native-run-finalizer.js"; +import { deliverExecutionStatuses } from "../execution-status-delivery.js"; import { commitNativeStatusDecision } from "./status-decision-committer.js"; import { NATIVE_STATUS_ARBITER_POLICY_VERSION, type NativeStatusDecision } from "./status-arbiter.js"; @@ -416,7 +417,7 @@ describeEmbeddedPostgres("native run finalizer / status decision committer — a }, ); - it("emits zero events when a reconciliation replay commits the same failed terminal result again", async () => { + it.each(["in_progress", "cancelled"])("does not redeliver a committed failure during reconciliation of a %s task", async (issueStatus) => { const fixture = await seedNativeRun(); await driveToCompleteResult(fixture, { ...CONTROL_PLANE_CONFORMANCE_TERMINAL, @@ -450,24 +451,63 @@ describeEmbeddedPostgres("native run finalizer / status decision committer — a runStatus: "failed", }); + await db.update(issues).set({ status: issueStatus }).where(eq(issues.id, fixture.issueId)); + const publish = vi.fn(); + await deliverExecutionStatuses(db, { publish }); + expect(publish.mock.calls.filter(([event]) => event.payload.runId === fixture.runId)).toHaveLength(1); + const [beforeReplay] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(beforeReplay.executionStatusDeliveryId).toBeNull(); + publish.mockClear(); + const callsBefore = mockTelemetryClient.track.mock.calls.length; const captureCallsBeforeReplay = mockCaptureRunFailure.mock.calls.length; - // The coordinator is already "committed" with a "failed" terminal - // result, and the run row is already "failed". Like "succeeded", - // "failed" sits inside projectCommittedRun's WHERE clause, so its write - // matches the row. The write changes nothing (failed -> failed), so it - // must not emit a second event for the same committed result, and it - // must not report a second Sentry event either. - await finalizeNativeRun({ - db, - runId: fixture.runId, - workspaceFinalizeStatus: "succeeded", - projectRunStatus: true, + // Simulate several periodic sweeps, including overlapping reconciliation. + for (let sweep = 0; sweep < 3; sweep += 1) { + await Promise.all([0, 1].map(() => finalizeNativeRun({ + db, + runId: fixture.runId, + workspaceFinalizeStatus: "succeeded", + projectRunStatus: true, + }))); + await deliverExecutionStatuses(db, { publish }); + } + expect(publish.mock.calls.filter(([event]) => event.payload.runId === fixture.runId)).toHaveLength(0); + const [afterReplay] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(afterReplay).toMatchObject({ + executionStatusDeliveryId: null, + updatedAt: beforeReplay.updatedAt, + nativePhaseUpdatedAt: beforeReplay.nativePhaseUpdatedAt, + finishedAt: beforeReplay.finishedAt, }); expect(agentTaskRunCalls(callsBefore)).toHaveLength(0); expect(captureRunFailureCallsFrom(captureCallsBeforeReplay)).toHaveLength(0); }); + it("repairs an incomplete committed failure once and preserves its pending delivery on replay", async () => { + const fixture = await seedNativeRun(); + await driveToCompleteResult(fixture, { + ...CONTROL_PLANE_CONFORMANCE_TERMINAL, + turnTerminalState: "failed", + runTerminalState: "failed", + }); + const finalize = () => finalizeNativeRun({ db, runId: fixture.runId, workspaceFinalizeStatus: "succeeded", projectRunStatus: true }); + await finalize(); + await db.update(heartbeatRuns).set({ + finishedAt: null, + nativePhase: "arbitrating", + executionStatusDeliveryId: null, + }).where(eq(heartbeatRuns.id, fixture.runId)); + await finalize(); + const [repaired] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(repaired.status).toBe("failed"); + expect(repaired.nativePhase).toBe("committed"); + expect(repaired.finishedAt).toBeInstanceOf(Date); + expect(repaired.executionStatusDeliveryId).toBeTypeOf("string"); + await finalize(); + const [replayed] = await db.select().from(heartbeatRuns).where(eq(heartbeatRuns.id, fixture.runId)); + expect(replayed).toEqual(repaired); + }); + it("emits zero events when a retryable-failure write's conditional status spread is omitted", async () => { const fixture = await seedNativeRun(); // recordNativeFinalizationFailure only needs the run and coordinator rows diff --git a/server/src/services/native-runtime/native-run-finalizer.ts b/server/src/services/native-runtime/native-run-finalizer.ts index 48c8ff358ae..8dfc177aef1 100644 --- a/server/src/services/native-runtime/native-run-finalizer.ts +++ b/server/src/services/native-runtime/native-run-finalizer.ts @@ -569,14 +569,15 @@ async function projectCommittedRun(input: { if (!["succeeded", "failed", "cancelled"].includes(String(terminalState))) { throw new Error("native_finalization_invalid"); } + const projectedStatus = projectNativeTerminalRunStatus( + terminalState as "succeeded" | "failed" | "cancelled", + ); const now = new Date(); const [updatedRun] = await input.db .update(heartbeatRuns) .set({ executionStatusDeliveryId: randomUUID(), - status: projectNativeTerminalRunStatus( - terminalState as "succeeded" | "failed" | "cancelled", - ), + status: projectedStatus, finishedAt: input.run.finishedAt ?? now, nativePhase: "committed", nativePhaseUpdatedAt: now, @@ -619,17 +620,24 @@ async function projectCommittedRun(input: { ), ), ), + // Reconciliation revisits committed results periodically. Only repair + // a changed projection; rewriting an unchanged failed run would mint a + // fresh status delivery (and failure toast) on every sweep. Check the + // current row so concurrent replays cannot both queue the same repair. + or( + sql`${heartbeatRuns.status} is distinct from ${projectedStatus}`, + isNull(heartbeatRuns.finishedAt), + sql`${heartbeatRuns.nativePhase} is distinct from 'committed'`, + ...(terminalState === "succeeded" + ? [isNotNull(heartbeatRuns.error), isNotNull(heartbeatRuns.errorCode)] + : []), + ), nativeRunnerOwnershipNotHeldCondition(), ), ) .returning(); - // The WHERE clause above allows "failed" as a source status, so a run that - // failed before its coordinator committed can still pick up the committed - // terminal state. A later reconciliation replay can enter this same path - // and match that clause again even when the committed state is still - // "failed" — the write changes nothing. Emit only when the status - // genuinely changed, so a reconciliation replay never emits twice for one - // committed terminal result. + // Metadata repairs can preserve the terminal status. Only a genuine status + // transition should emit another terminal event. if (updatedRun && updatedRun.status !== input.run.status) { await emitAgentTaskRun(input.db, updatedRun); void reportRunFailure(input.db, updatedRun); diff --git a/ui/src/context/LiveUpdatesProvider.hook.test.tsx b/ui/src/context/LiveUpdatesProvider.hook.test.tsx index 41546fb5717..8650bfc355a 100644 --- a/ui/src/context/LiveUpdatesProvider.hook.test.tsx +++ b/ui/src/context/LiveUpdatesProvider.hook.test.tsx @@ -170,17 +170,64 @@ describe("LiveUpdatesProvider socket run notification scope", () => { vi.unstubAllGlobals(); }); - async function receiveStatus(payload: Record) { + async function receiveStatus(payload: Record, createdAt = "2026-09-09T18:00:00.000Z") { await reactAct(async () => { root!.render(Visible task); }); await vi.waitFor(() => expect(sockets).toHaveLength(1)); expect(sockets[0]?.onmessage).toBeTypeOf("function"); await reactAct(async () => sockets[0]!.onmessage!(new MessageEvent("message", { - data: JSON.stringify({ id: 1, companyId: "company-1", type: "heartbeat.run.status", createdAt: "2026-09-09T18:00:00.000Z", payload }), + data: JSON.stringify({ id: Date.now(), companyId: "company-1", type: "heartbeat.run.status", createdAt, payload }), }))); } + it("announces a run outcome once across minute-apart redeliveries but still refreshes caches", async () => { + const clock = vi.spyOn(Date, "now"); + const payload = { runId: "unrelated-run", agentId: "other-agent", status: "failed", finishedAt: "2026-09-09T18:00:00.000Z" }; + clock.mockReturnValue(Date.parse(payload.finishedAt)); + await receiveStatus({ ...payload, deliveryId: "first" }); + expect(pushToast).toHaveBeenCalledTimes(1); + queryClient.setQueryData(queryKeys.runDetail(payload.runId), { id: payload.runId, status: "running" }); + for (let minute = 1; minute <= 3; minute += 1) { + const now = Date.parse(payload.finishedAt) + minute * 60_000; + clock.mockReturnValue(now); + await receiveStatus({ ...payload, deliveryId: `retry-${minute}` }, new Date(now).toISOString()); + } + expect(pushToast).toHaveBeenCalledTimes(1); + expect(queryClient.getQueryData(queryKeys.runDetail(payload.runId))).toMatchObject({ status: "failed" }); + await receiveStatus({ ...payload, runId: "new-failure" }); + expect(pushToast).toHaveBeenCalledTimes(2); + }); + + it("refreshes an old terminal run without announcing its first delivery as a new failure", async () => { + queryClient.setQueryData(queryKeys.runDetail("old-run"), { id: "old-run", status: "running" }); + await receiveStatus({ + runId: "old-run", agentId: "other-agent", status: "failed", + finishedAt: "2026-09-07T18:00:00.000Z", deliveryId: "new-delivery-of-old-failure", + }); + expect(pushToast).not.toHaveBeenCalled(); + expect(queryClient.getQueryData(queryKeys.runDetail("old-run"))).toMatchObject({ status: "failed" }); + }); + + it("does not announce an outcome again after hiding and reopening the window", async () => { + const payload = { runId: "unrelated-run", agentId: "other-agent", status: "failed" }; + await receiveStatus(payload); + const visibility = vi.spyOn(document, "visibilityState", "get"); + await reactAct(async () => { + visibility.mockReturnValue("hidden"); + document.dispatchEvent(new Event("visibilitychange")); + }); + await reactAct(async () => { + visibility.mockReturnValue("visible"); + document.dispatchEvent(new Event("visibilitychange")); + }); + await vi.waitFor(() => expect(sockets).toHaveLength(2)); + await reactAct(async () => sockets[1]!.onmessage!(new MessageEvent("message", { + data: JSON.stringify({ id: 2, companyId: "company-1", type: "heartbeat.run.status", createdAt: "2026-09-09T18:01:00.000Z", payload }), + }))); + expect(pushToast).toHaveBeenCalledTimes(1); + }); + it("disconnects while hidden and reconciles active queries once on return", async () => { await receiveStatus({ runId: "child-run", agentId: "child-agent", status: "running" }); const invalidate = vi.spyOn(queryClient, "invalidateQueries"); diff --git a/ui/src/context/LiveUpdatesProvider.tsx b/ui/src/context/LiveUpdatesProvider.tsx index 8ea8c42dd0d..f86c71b47b7 100644 --- a/ui/src/context/LiveUpdatesProvider.tsx +++ b/ui/src/context/LiveUpdatesProvider.tsx @@ -54,6 +54,8 @@ import { tryCreateWebSocket } from "../lib/websocket"; const TOAST_COOLDOWN_WINDOW_MS = 10_000; const TOAST_COOLDOWN_MAX = 3; const RECONNECT_SUPPRESS_MS = 2000; +const RUN_TOAST_MAX_AGE_MS = 5 * 60_000; +const MAX_OBSERVED_RUN_OUTCOMES = 2000; const DISCONNECTED_POLL_INTERVAL_MS = 15_000; const SOCKET_CONNECTING = 0; const SOCKET_OPEN = 1; @@ -1587,6 +1589,29 @@ function invalidateActivityQueries( interface ToastGate { cooldownHits: Map; suppressUntil: number; + observedRunOutcomes: Set; +} + +function observeRunOutcome(gate: ToastGate, event: LiveEvent): boolean { + const payload = event.payload ?? {}; + const runId = readString(payload.runId); + const status = readString(payload.status); + if (!runId || !status || !TERMINAL_RUN_STATUSES.has(status)) return false; + const key = `${event.companyId}:${runId}:${status}`; + const observed = gate.observedRunOutcomes.has(key); + // Refresh insertion order so repeated deliveries stay remembered even in a + // busy company. Remember suppressed outcomes too (visible task/reconnect). + gate.observedRunOutcomes.delete(key); + gate.observedRunOutcomes.add(key); + if (gate.observedRunOutcomes.size > MAX_OBSERVED_RUN_OUTCOMES) { + gate.observedRunOutcomes.delete(gate.observedRunOutcomes.values().next().value!); + } + // Use the two server timestamps: delivery time is not failure time. This + // also keeps historical failures quiet after a reload, without client clock + // skew hiding fresh failures. Missing timestamps retain legacy behavior. + const finishedAt = Date.parse(readString(payload.finishedAt) ?? ""); + const deliveredAt = Date.parse(event.createdAt); + return observed || deliveredAt - finishedAt > RUN_TOAST_MAX_AGE_MS; } function shouldSuppressToast(gate: ToastGate, category: string): boolean { @@ -1636,7 +1661,8 @@ function handleLiveEvent( // Resolve membership before terminal lifecycle patches remove live-run rows. const suppressRunToast = event.type === "heartbeat.run.status" && - shouldSuppressRunStatusToastForVisibleIssue(queryClient, pathname, payload); + (observeRunOutcome(gate, event) || + shouldSuppressRunStatusToastForVisibleIssue(queryClient, pathname, payload)); const liveStatusPatch = readRunLiveStatusPatchFromPayload( payload, event.createdAt, @@ -1837,6 +1863,7 @@ export function LiveUpdatesProvider({ children }: { children: ReactNode }) { const gateRef = useRef({ cooldownHits: new Map(), suppressUntil: 0, + observedRunOutcomes: new Set(), }); const pathnameRef = useRef(location.pathname); const { data: session, status: sessionStatus } = useQuery({