diff --git a/apps/mobile/src/features/threads/ActionResumeNotice.tsx b/apps/mobile/src/features/threads/ActionResumeNotice.tsx index 00ce66397a2e..a59bbf632f24 100644 --- a/apps/mobile/src/features/threads/ActionResumeNotice.tsx +++ b/apps/mobile/src/features/threads/ActionResumeNotice.tsx @@ -1,4 +1,5 @@ import type { EnvironmentId, OrchestrationThreadShell } from "@t3tools/contracts"; +import { actionRunningPresentation } from "@t3tools/shared/actionResume"; import * as Cause from "effect/Cause"; import { useCallback, useEffect, useState } from "react"; import { Alert, View } from "react-native"; @@ -71,16 +72,41 @@ export function ActionResumeNotice(props: { ); if (action?.outcome === "running") { + const presentation = actionRunningPresentation(action); + const working = presentation.state === "working"; return ( - + - + - - Waiting for {action.actionName} + + {presentation.label}: {action.actionName} - - The agent will resume once this Action finishes and the thread is idle. + + {presentation.summary} void; }) { - const status = props.exitCode ?? props.validatedStatus; + const tone = + props.outcome === "success" + ? { + container: "border-emerald-500/25 bg-emerald-500/[0.06]", + text: "text-adaptive-emerald-700-300", + color: "#30d158", + icon: "checkmark.circle" as const, + } + : props.outcome === "error" + ? { + container: "border-rose-500/25 bg-rose-500/[0.06]", + text: "text-adaptive-rose-700-300", + color: "#ff453a", + icon: "xmark.circle.fill" as const, + } + : props.outcome === "cancelled" + ? { + container: "border-adaptive-black-a10-white-a10 bg-neutral-500/[0.06]", + text: "text-foreground-muted", + color: props.iconColor, + icon: "xmark.circle.fill" as const, + } + : props.outcome === "blocked" + ? { + container: "border-violet-500/25 bg-violet-500/[0.06]", + text: "text-adaptive-violet-700-300", + color: "#bf5af2", + icon: "exclamationmark.triangle" as const, + } + : { + container: "border-amber-500/25 bg-amber-500/[0.06]", + text: "text-adaptive-amber-700-300", + color: "#eab308", + icon: "exclamationmark.triangle" as const, + }; const heading = ( <> - - - Action completed: {props.actionName} Status: {status} + + + {props.outcomeLabel}: {props.actionName} ); return ( - + {props.detailedOutputAvailable ? ( {heading} @@ -1238,7 +1274,7 @@ const ActionFollowUpCard = memo(function ActionFollowUpCard(props: { @@ -1263,7 +1299,7 @@ const ActionFollowUpCard = memo(function ActionFollowUpCard(props: { ) : ( - {props.lastOutputLine} + {props.summary} {props.detailedOutputAvailable ? ( diff --git a/apps/mobile/src/features/threads/thread-list-items.tsx b/apps/mobile/src/features/threads/thread-list-items.tsx index 028ebbc8ffcc..024e02d2ea0c 100644 --- a/apps/mobile/src/features/threads/thread-list-items.tsx +++ b/apps/mobile/src/features/threads/thread-list-items.tsx @@ -30,6 +30,7 @@ import type { HomeGroupDisplayAction } from "../home/homeListItems"; import { ThreadSwipeable } from "../home/thread-swipe-actions"; import { buildThreadTitleRegenerationMenuItems } from "./thread-title-regeneration-menu"; import { resolveThreadStatus, shouldShowActionWaitingIndicator } from "./threadPresentation"; +import { actionRunningPresentation } from "@t3tools/shared/actionResume"; import { ThreadSearchMatchExcerpt } from "./thread-search-match"; /** @@ -480,7 +481,9 @@ export const ThreadListRow = memo(function ThreadListRow(props: { const showActionWaitingIndicator = shouldShowActionWaitingIndicator(thread, status?.kind ?? null); const threadAccessibilityLabel = [ thread.title, - showActionWaitingIndicator && runningAction ? `Waiting for ${runningAction.actionName}` : null, + showActionWaitingIndicator && runningAction + ? `Waiting for ${runningAction.actionName}. ${actionRunningPresentation(runningAction).summary}` + : null, pr?.accessibilityLabel ?? null, ] .filter((part): part is string => part !== null) @@ -623,7 +626,7 @@ export const ThreadListRow = memo(function ThreadListRow(props: { const actionStatusIndicator = showActionWaitingIndicator && runningAction ? ( diff --git a/apps/mobile/src/features/threads/thread-list-v2-items.tsx b/apps/mobile/src/features/threads/thread-list-v2-items.tsx index 8a50d23af93e..51f00452025e 100644 --- a/apps/mobile/src/features/threads/thread-list-v2-items.tsx +++ b/apps/mobile/src/features/threads/thread-list-v2-items.tsx @@ -30,6 +30,7 @@ import { resolveWorktreeCleanupStatus, shouldShowActionWaitingIndicator, } from "./threadPresentation"; +import { actionRunningPresentation } from "@t3tools/shared/actionResume"; import { resolveThreadListV2ChangeRequestState, resolveThreadListV2CleanupActions, @@ -460,7 +461,9 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { const timeLabel = threadTimeLabel(thread); const threadAccessibilityLabel = [ thread.title, - showActionWaitingIndicator && runningAction ? `Waiting for ${runningAction.actionName}` : null, + showActionWaitingIndicator && runningAction + ? `Waiting for ${runningAction.actionName}. ${actionRunningPresentation(runningAction).summary}` + : null, ] .filter((part): part is string => part !== null) .join(", "); @@ -833,7 +836,7 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { {showActionWaitingIndicator && runningAction ? ( { expect(shouldShowActionWaitingIndicator(working, "working")).toBe(true); }); + it("presents Action working progress without a redundant secondary indicator", () => { + const actionResume = { + outcome: "running", + actionName: "QA", + progress: { + version: 1, + state: "working", + summary: "Running checks", + updatedAt: NOW, + }, + } as NonNullable; + const working = makeThread({ + id: ThreadId.make("working-progress"), + title: "Working Action", + actionResume, + }); + + expect(resolveThreadListV2Status(working)).toBe("working"); + expect(resolveThreadStatus(working)).toMatchObject({ kind: "working", label: "Working" }); + expect(shouldShowActionWaitingIndicator(working, "working")).toBe(false); + }); + it("resolves ready for quiescent threads", () => { expect(resolveThreadListV2Status(makeThread({ id: ThreadId.make("t"), title: "t" }))).toBe( "ready", diff --git a/apps/mobile/src/features/threads/threadListV2.ts b/apps/mobile/src/features/threads/threadListV2.ts index 57dc0bb136f1..c6cd172df6cd 100644 --- a/apps/mobile/src/features/threads/threadListV2.ts +++ b/apps/mobile/src/features/threads/threadListV2.ts @@ -17,6 +17,7 @@ import { sortPinnedThreadsByOrderKey, } from "@t3tools/client-runtime/state/thread-sort"; import type { EnvironmentId, ProjectId, ThreadLinkedPullRequest } from "@t3tools/contracts"; +import { actionRunningPresentation } from "@t3tools/shared/actionResume"; import type { PendingNewTask } from "../../state/use-pending-new-tasks"; @@ -185,7 +186,7 @@ export function resolveThreadListV2Status( return "failed"; } if (thread.actionResume?.outcome === "running") { - return "waiting"; + return actionRunningPresentation(thread.actionResume).state; } return "ready"; } diff --git a/apps/mobile/src/features/threads/threadPresentation.ts b/apps/mobile/src/features/threads/threadPresentation.ts index 38183c6c477b..a2ad2ff22a39 100644 --- a/apps/mobile/src/features/threads/threadPresentation.ts +++ b/apps/mobile/src/features/threads/threadPresentation.ts @@ -1,5 +1,6 @@ import type { StatusTone } from "../../components/StatusPill"; import type { OrchestrationLatestTurn, OrchestrationSession } from "@t3tools/contracts"; +import { actionRunningPresentation } from "@t3tools/shared/actionResume"; import { EnvironmentThreadShell } from "@t3tools/client-runtime/state/shell"; export function threadSortValue(thread: EnvironmentThreadShell): number { @@ -39,7 +40,11 @@ export function shouldShowActionWaitingIndicator( thread: Pick, primaryStatus: string | null, ): boolean { - return thread.actionResume?.outcome === "running" && primaryStatus !== "waiting"; + return ( + thread.actionResume?.outcome === "running" && + actionRunningPresentation(thread.actionResume).state === "waiting" && + primaryStatus !== "waiting" + ); } function isLatestTurnSettled( @@ -157,13 +162,18 @@ export function resolveThreadStatus( } if (thread.actionResume?.outcome === "running") { + const action = actionRunningPresentation(thread.actionResume); return { - kind: "waiting", - label: "Waiting", - pillClassName: "bg-adaptive-yellow-500-a12-a16", - textClassName: "text-adaptive-yellow-700-300", - iconColor: "#eab308", - iconBackground: "rgba(234,179,8,0.22)", + kind: action.state, + label: action.label, + pillClassName: + action.state === "working" + ? "bg-adaptive-sky-500-a12-a16" + : "bg-adaptive-yellow-500-a12-a16", + textClassName: + action.state === "working" ? "text-adaptive-sky-700-300" : "text-adaptive-yellow-700-300", + iconColor: action.state === "working" ? "#0a84ff" : "#eab308", + iconBackground: action.state === "working" ? "rgba(10,132,255,0.22)" : "rgba(234,179,8,0.22)", pulse: false, }; } diff --git a/apps/server/src/actionResume/ActionResume.test.ts b/apps/server/src/actionResume/ActionResume.test.ts index 65037f735bcd..c51606f77fa6 100644 --- a/apps/server/src/actionResume/ActionResume.test.ts +++ b/apps/server/src/actionResume/ActionResume.test.ts @@ -1,6 +1,7 @@ import * as NodeServices from "@effect/platform-node/NodeServices"; import { assert, it } from "@effect/vitest"; import { + type ActionProgress, type ActionResumeState, EventId, ProjectId, @@ -19,6 +20,7 @@ import * as Deferred from "effect/Deferred"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Stream from "effect/Stream"; +import * as TestClock from "effect/testing/TestClock"; import { ACTION_EVENT_TOKEN_ENV, ACTION_RUN_ID_ENV, @@ -123,15 +125,24 @@ it.effect("runs one opted-in Action and delivers exactly one automated follow-up let terminalStatus: "running" | "exited" = "running"; let failWrite = false; let admissionClosed = false; + let progressFlushed: Deferred.Deferred | undefined; const admittedKinds: string[] = []; let terminalListener: ((event: TerminalEvent) => Effect.Effect) | undefined; const dependencies = Layer.mergeAll( Layer.mock(OrchestrationEngineService)({ dispatch: (command) => - Effect.sync(() => { + Effect.gen(function* () { timeline.push(`dispatch:${command.type}`); dispatched.push(command); + if ( + progressFlushed !== undefined && + command.type === "thread.activity.append" && + command.activity.kind === ActionResume.ACTION_RESUME_ACTIVITY_KIND && + (command.activity.payload as ActionResumeState).progress?.phase === "review" + ) { + yield* Deferred.succeed(progressFlushed, undefined); + } return { sequence: dispatched.length }; }), streamDomainEvents: Stream.never, @@ -228,7 +239,10 @@ it.effect("runs one opted-in Action and delivers exactly one automated follow-up ); return Effect.gen(function* () { + progressFlushed = yield* Deferred.make(); const service = yield* ActionResume.ActionResume; + // Let startup reconciliation settle while the registry is still empty. + yield* Effect.yieldNow; const listed = yield* service.listProjectActions({ threadId, providerInstanceId }); assert.deepEqual( listed.map(({ id, resumeEligible }) => ({ id, resumeEligible })), @@ -294,6 +308,16 @@ it.effect("runs one opted-in Action and delivers exactly one automated follow-up assert.isDefined(terminalListener); const startMarker = ActionResume.actionOutputMarker(running.runId, "start"); const endMarker = ActionResume.actionOutputMarker(running.runId, "end"); + const progressFrame = ( + state: "working" | "waiting", + summary: string, + fields: Partial> = {}, + ) => + actionProtocolFrame({ + runId: running.runId, + token: eventToken!, + event: { kind: "progress", progress: { version: 1, state, summary, ...fields } }, + }); const resultFrame = actionProtocolFrame({ runId: running.runId, token: eventToken!, @@ -317,8 +341,105 @@ it.effect("runs one opted-in Action and delivers exactly one automated follow-up type: "output", threadId, terminalId: running.terminalId, - data: `${startMarker.slice(-2)}QA failed: \u001b[31mexpected 2, received 3\u001b[0m\n${resultFrame}${endMarker}prompt`, + data: `${startMarker.slice(-2)}QA failed: \u001b[31mexpected 2, received 3\u001b[0m\n${progressFrame("working", "Running checks")}${progressFrame("working", "Running checks")}${progressFrame("working", "Running tests")}${progressFrame("waiting", "Waiting for review", { phase: "ci", current: 1, total: 3, unit: "check" })}${resultFrame}${endMarker}prompt`, + }); + const registry = yield* ThreadActionResume.ThreadActionResumeService; + assert.deepInclude(registry.getLatest(threadId), { + outcome: "running", + revision: 2, }); + assert.equal(registry.getLatest(threadId)?.progress?.state, "waiting"); + assert.equal(registry.getLatest(threadId)?.progress?.summary, "Waiting for review"); + assert.deepEqual( + dispatched.flatMap((command) => { + if ( + command.type !== "thread.activity.append" || + command.activity.kind !== ActionResume.ACTION_RESUME_ACTIVITY_KIND + ) { + return []; + } + const payload = command.activity.payload as ActionResumeState; + return payload.runId === running.runId && payload.outcome === "running" + ? [payload.revision] + : []; + }), + [0, 1, 2], + ); + yield* terminalListener!({ + type: "output", + threadId, + terminalId: running.terminalId, + data: progressFrame("waiting", "Waiting for review", { + phase: "review", + detail: "Review is still pending", + current: 2, + total: 3, + unit: "check", + }), + }); + yield* TestClock.adjust(999); + assert.equal(registry.getLatest(threadId)?.progress?.summary, "Waiting for review"); + yield* TestClock.adjust(1); + yield* Deferred.await(progressFlushed); + assert.deepInclude(registry.getLatest(threadId)?.progress, { + state: "waiting", + summary: "Waiting for review", + phase: "review", + detail: "Review is still pending", + current: 2, + total: 3, + unit: "check", + }); + assert.deepEqual( + dispatched.flatMap((command) => { + if ( + command.type !== "thread.activity.append" || + command.activity.kind !== ActionResume.ACTION_RESUME_ACTIVITY_KIND + ) { + return []; + } + const payload = command.activity.payload as ActionResumeState; + return payload.runId === running.runId && payload.outcome === "running" + ? [payload.revision] + : []; + }), + [0, 1, 2, 3], + ); + yield* terminalListener!({ + type: "output", + threadId, + terminalId: running.terminalId, + data: `${progressFrame("waiting", "Waiting for review", { + phase: "approval", + current: 3, + total: 3, + unit: "check", + })}${progressFrame("waiting", "Waiting for review", { + phase: "review", + detail: "Review is still pending", + current: 2, + total: 3, + unit: "check", + })}`, + }); + yield* TestClock.adjust(1_000); + yield* Effect.yieldNow; + assert.equal(registry.getLatest(threadId)?.revision, 3); + assert.equal(registry.getLatest(threadId)?.progress?.phase, "review"); + assert.deepEqual( + [ + ...new Set( + dispatched.flatMap((command) => + command.type === "thread.activity.append" && + command.activity.kind === ActionResume.ACTION_RESUME_ACTIVITY_KIND && + (command.activity.payload as ActionResumeState).runId === running.runId + ? [command.activity.id] + : [], + ), + ), + ], + [`action-resume:${running.runId}`], + ); yield* terminalListener!({ type: "exited", threadId, @@ -349,7 +470,6 @@ it.effect("runs one opted-in Action and delivers exactly one automated follow-up assert.equal(turnStarts[0]?.runtimeMode, thread.runtimeMode); assert.equal(turnStarts[0]?.interactionMode, thread.interactionMode); - const registry = yield* ThreadActionResume.ThreadActionResumeService; assert.deepInclude(registry.getLatest(threadId), { outcome: "succeeded", delivery: "delivered", @@ -437,7 +557,12 @@ it.effect("runs one opted-in Action and delivers exactly one automated follow-up outcome: "succeeded", delivery: "delivered", }); - }).pipe(Effect.provide(ActionResume.layer.pipe(Layer.provideMerge(dependencies))), Effect.scoped); + }).pipe( + Effect.provide( + ActionResume.layer.pipe(Layer.provideMerge(Layer.merge(dependencies, TestClock.layer()))), + ), + Effect.scoped, + ); }); it.effect("requires an explicit resume after a running Action is found on startup", () => diff --git a/apps/server/src/actionResume/ActionResume.ts b/apps/server/src/actionResume/ActionResume.ts index 74b795c2996b..4b419bfbc950 100644 --- a/apps/server/src/actionResume/ActionResume.ts +++ b/apps/server/src/actionResume/ActionResume.ts @@ -15,6 +15,7 @@ import { EventId, MessageId, ProviderDriverKind, + type ActionProgress, type ProviderInstanceId, type ProjectScript, type ThreadId, @@ -28,9 +29,11 @@ import { type ActionProtocolDecoder, } from "@t3tools/shared/actionResumeProtocol"; import * as Cause from "effect/Cause"; +import * as Clock from "effect/Clock"; import * as Context from "effect/Context"; import * as Crypto from "effect/Crypto"; import * as DateTime from "effect/DateTime"; +import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; @@ -77,8 +80,25 @@ interface FinishActionInput { interface ActionProtocolCapture { readonly decoder: ActionProtocolDecoder; report?: ActionResumeState["report"]; + lastObservedProgress?: ActionProgress; + lastAcceptedProgressAtMs?: number; + acceptedProgressCount: number; + pendingProgress?: ActionProgress; + progressFlushGeneration: number; + progressFlushScheduled?: boolean; + progressLimitWarned?: boolean; } +const progressEquals = (left: ActionProgress | undefined, right: ActionProgress) => + left?.version === right.version && + left.state === right.state && + left.summary === right.summary && + left.phase === right.phase && + left.detail === right.detail && + left.current === right.current && + left.total === right.total && + left.unit === right.unit; + export class ActionResume extends Context.Service< ActionResume, { @@ -107,7 +127,9 @@ const nowIso = Effect.map(DateTime.now, DateTime.formatIso); const outcomeSummary = (state: ActionResumeState): string => { switch (state.outcome) { case "running": - return `Waiting for Action: ${state.actionName}`; + return state.progress === undefined + ? `Waiting for Action: ${state.actionName}` + : `${state.progress.state === "working" ? "Working" : "Waiting"}: ${state.progress.summary}`; case "succeeded": return `Action completed: ${state.actionName}`; case "failed": @@ -127,6 +149,8 @@ const outcomeTone = (state: ActionResumeState): "info" | "error" => state.outcome === "failed" || state.outcome === "process_lost" ? "error" : "info"; const MAX_ACTION_OUTPUT_CHARS = 12_000; +const ACTION_PROGRESS_MIN_INTERVAL_MS = 1_000; +const ACTION_PROGRESS_MAX_UPDATES = 128; const ACTION_OUTPUT_OSC = "777;T3ActionOutput"; type ActionOutputBoundary = "start" | "end"; @@ -226,6 +250,7 @@ const followUpText = (state: ActionResumeState, outputTail: string | undefined): actionId: state.actionId, runId: state.runId, validatedStatus: status, + lifecycleOutcome: state.outcome, exitCode: state.exitCode, report: state.report, output: outputTail, @@ -279,6 +304,8 @@ const mapActionResumeError = ); const make = Effect.gen(function* () { + const serviceScope = yield* Effect.scope; + const clock = yield* Clock.Clock; const crypto = yield* Crypto.Crypto; const engine = yield* OrchestrationEngineService; const snapshots = yield* ProjectionSnapshotQuery; @@ -301,15 +328,19 @@ const make = Effect.gen(function* () { }, ); - const persistState = Effect.fn("ActionResume.persistState")(function* (state: ActionResumeState) { - const previous = registry.getLatest(state.threadId); + const persistState = Effect.fn("ActionResume.persistState")(function* (input: ActionResumeState) { + const previous = registry.getLatest(input.threadId); + const state: ActionResumeState = { + ...input, + revision: + previous?.runId === input.runId ? (previous.revision ?? 0) + 1 : (input.revision ?? 0), + }; registry.record(state); - const activityId = EventId.make( - `action-resume:${state.runId}:${state.outcome}:${state.delivery}`, - ); + const activityId = EventId.make(`action-resume:${state.runId}`); const commandId = CommandId.make( - `server:action-resume:${state.runId}:${state.outcome}:${state.delivery}`, + `server:action-resume:${state.runId}:${state.revision}:${state.outcome}:${state.delivery}`, ); + const activityCreatedAt = state.finishedAt ?? state.progress?.updatedAt ?? state.startedAt; yield* engine .dispatch({ type: "thread.activity.append", @@ -322,9 +353,9 @@ const make = Effect.gen(function* () { summary: outcomeSummary(state), payload: state, turnId: null, - createdAt: state.finishedAt ?? state.startedAt, + createdAt: activityCreatedAt, }, - createdAt: state.finishedAt ?? state.startedAt, + createdAt: activityCreatedAt, }) .pipe( Effect.catchCause((cause) => { @@ -333,6 +364,109 @@ const make = Effect.gen(function* () { return Effect.failCause(cause); }), ); + return state; + }); + + const flushPendingProgressUnlocked = Effect.fn("ActionResume.flushPendingProgressUnlocked")( + function* (threadId: ThreadId, runId: string, generation: number) { + const protocol = protocolCaptureByRunId.get(runId); + if (protocol === undefined || protocol.progressFlushGeneration !== generation) return; + + protocol.progressFlushScheduled = false; + const pending = protocol.pendingProgress; + const current = registry.getLatest(threadId); + if (pending === undefined || current?.runId !== runId || current.outcome !== "running") + return; + + const updatedAt = yield* nowIso; + yield* persistState({ ...current, progress: { ...pending, updatedAt } }); + protocol.lastObservedProgress = pending; + protocol.lastAcceptedProgressAtMs = yield* clock.currentTimeMillis; + protocol.acceptedProgressCount += 1; + delete protocol.pendingProgress; + }, + ); + + const acceptProgressUnlocked = Effect.fn("ActionResume.acceptProgressUnlocked")(function* ( + threadId: ThreadId, + runId: string, + progress: ActionProgress, + ) { + const current = registry.getLatest(threadId); + const protocol = protocolCaptureByRunId.get(runId); + if (current?.runId !== runId || current.outcome !== "running" || protocol === undefined) return; + + if (progressEquals(current.progress, progress)) { + protocol.lastObservedProgress = progress; + if (protocol.pendingProgress !== undefined) { + protocol.progressFlushGeneration += 1; + protocol.progressFlushScheduled = false; + delete protocol.pendingProgress; + } + return; + } + + if ( + progressEquals(protocol.lastObservedProgress, progress) && + protocol.pendingProgress === undefined + ) { + return; + } + + if (protocol.acceptedProgressCount >= ACTION_PROGRESS_MAX_UPDATES) { + protocol.lastObservedProgress = progress; + if (protocol.progressLimitWarned !== true) { + protocol.progressLimitWarned = true; + yield* Effect.logWarning("Action progress update limit reached", { + threadId, + runId, + limit: ACTION_PROGRESS_MAX_UPDATES, + }); + } + return; + } + + const acceptedAtMs = yield* clock.currentTimeMillis; + const stateChanged = current.progress?.state !== progress.state; + if ( + !stateChanged && + protocol.lastAcceptedProgressAtMs !== undefined && + acceptedAtMs - protocol.lastAcceptedProgressAtMs < ACTION_PROGRESS_MIN_INTERVAL_MS + ) { + protocol.lastObservedProgress = progress; + protocol.pendingProgress = progress; + if (protocol.progressFlushScheduled !== true) { + protocol.progressFlushScheduled = true; + const generation = ++protocol.progressFlushGeneration; + const remainingMs = + ACTION_PROGRESS_MIN_INTERVAL_MS - (acceptedAtMs - protocol.lastAcceptedProgressAtMs); + yield* clock.sleep(Duration.millis(remainingMs)).pipe( + Effect.andThen( + mutex.withPermits(1)(flushPendingProgressUnlocked(threadId, runId, generation)), + ), + Effect.catchCause((cause) => + Cause.hasInterrupts(cause) + ? Effect.void + : Effect.logWarning("Could not flush deferred Action progress", { + threadId, + runId, + cause: Cause.pretty(cause), + }), + ), + Effect.forkIn(serviceScope, { startImmediately: true }), + ); + } + return; + } + + const updatedAt = yield* nowIso; + yield* persistState({ ...current, progress: { ...progress, updatedAt } }); + protocol.progressFlushGeneration += 1; + protocol.progressFlushScheduled = false; + protocol.lastObservedProgress = progress; + protocol.lastAcceptedProgressAtMs = acceptedAtMs; + protocol.acceptedProgressCount += 1; + delete protocol.pendingProgress; }); const eligibleThreadForFollowUp = Effect.fn("ActionResume.eligibleThreadForFollowUp")(function* ( @@ -431,8 +565,12 @@ const make = Effect.gen(function* () { (input.outcome === "succeeded" || input.outcome === "failed" || input.outcome === "cancelled_by_user"); - const report = - input.outcome === "succeeded" ? protocolCaptureByRunId.get(current.runId)?.report : undefined; + const protocol = protocolCaptureByRunId.get(current.runId); + const report = input.outcome === "succeeded" ? protocol?.report : undefined; + const progress = + protocol?.pendingProgress === undefined + ? current.progress + : { ...protocol.pendingProgress, updatedAt: finishedAt }; const next: ActionResumeState = { ...current, outcome: input.outcome, @@ -444,6 +582,7 @@ const make = Effect.gen(function* () { finishedAt, exitCode: input.exitCode ?? null, exitSignal: input.exitSignal ?? null, + ...(progress === undefined ? {} : { progress }), ...(report === undefined ? {} : { report }), }; yield* persistState(next); @@ -579,6 +718,7 @@ const make = Effect.gen(function* () { finishedAt: null, exitCode: null, exitSignal: null, + revision: 0, }; const cwd = thread.worktreePath ?? project.workspaceRoot; const env = projectScriptRuntimeEnv({ @@ -606,6 +746,8 @@ const make = Effect.gen(function* () { outputCaptureByRunId.set(runId, createActionOutputCapture(runId)); protocolCaptureByRunId.set(runId, { decoder: createActionProtocolDecoder({ runId, token: eventToken }), + acceptedProgressCount: 0, + progressFlushGeneration: 0, }); yield* terminals.write({ threadId: invocation.threadId, @@ -762,8 +904,21 @@ const make = Effect.gen(function* () { const decoded = protocol.decoder.push(event.data); consumeActionTerminalOutput(capture, decoded.output); for (const actionEvent of decoded.events) { - if (actionEvent.kind !== "result") continue; - if (protocol.report === undefined) protocol.report = actionEvent.report; + if (actionEvent.kind === "progress") { + yield* mutex + .withPermits(1)( + acceptProgressUnlocked(state.threadId, state.runId, actionEvent.progress), + ) + .pipe( + Effect.catchCause((cause) => + Effect.logWarning("Could not persist Action progress", { + threadId: state.threadId, + runId: state.runId, + cause: Cause.pretty(cause), + }), + ), + ); + } else if (protocol.report === undefined) protocol.report = actionEvent.report; else { yield* Effect.logWarning("Action emitted more than one terminal result", { threadId: state.threadId, diff --git a/apps/server/src/orchestration/ThreadActionResume.test.ts b/apps/server/src/orchestration/ThreadActionResume.test.ts index b96c6dd791b9..9617565a4879 100644 --- a/apps/server/src/orchestration/ThreadActionResume.test.ts +++ b/apps/server/src/orchestration/ThreadActionResume.test.ts @@ -59,3 +59,42 @@ it("keeps a genuinely pending newer Action visible after hydration", () => { delivery: "pending", }); }); + +it("hydrates the latest progress revision for one running Action", () => { + const registry = make(); + + registry.hydrate([ + state({ + outcome: "running", + delivery: "armed", + finishedAt: null, + exitCode: null, + revision: 2, + progress: { + version: 1, + state: "working", + summary: "Running checks", + updatedAt: "2026-08-17T00:00:10.000Z", + }, + }), + state({ + outcome: "running", + delivery: "armed", + finishedAt: null, + exitCode: null, + revision: 3, + progress: { + version: 1, + state: "waiting", + summary: "Waiting for review", + updatedAt: "2026-08-17T00:00:20.000Z", + }, + }), + ]); + + assert.deepInclude(registry.getForShell(threadId), { + revision: 3, + }); + assert.equal(registry.getForShell(threadId)?.progress?.state, "waiting"); + assert.equal(registry.getForShell(threadId)?.progress?.summary, "Waiting for review"); +}); diff --git a/apps/server/src/orchestration/ThreadActionResume.ts b/apps/server/src/orchestration/ThreadActionResume.ts index 2ce9da3e065f..f14211454e5e 100644 --- a/apps/server/src/orchestration/ThreadActionResume.ts +++ b/apps/server/src/orchestration/ThreadActionResume.ts @@ -31,6 +31,8 @@ const hydrationDeliveryRank = (delivery: ActionResumeState["delivery"]): number } }; +const hydrationRevision = (state: ActionResumeState): number => state.revision ?? 0; + export interface ThreadActionResumeShape { /** Restore one latest run per thread without reviving superseded delivery states. */ readonly hydrate: (states: ReadonlyArray) => void; @@ -53,7 +55,9 @@ export function make(): ThreadActionResumeShape { current === undefined || state.startedAt > current.startedAt || (state.runId === current.runId && - hydrationDeliveryRank(state.delivery) > hydrationDeliveryRank(current.delivery)) + (hydrationRevision(state) > hydrationRevision(current) || + (hydrationRevision(state) === hydrationRevision(current) && + hydrationDeliveryRank(state.delivery) > hydrationDeliveryRank(current.delivery)))) ) { latestByThreadId.set(state.threadId, state); } diff --git a/apps/web/src/components/LegacySidebar.tsx b/apps/web/src/components/LegacySidebar.tsx index 9b11e4ab910f..d40cda858113 100644 --- a/apps/web/src/components/LegacySidebar.tsx +++ b/apps/web/src/components/LegacySidebar.tsx @@ -24,6 +24,7 @@ import { ProjectFavicon } from "./ProjectFavicon"; import { SidebarDraftBlock } from "./Sidebar"; import { useAtomValue } from "@effect/atom-react"; import { autoAnimate } from "@formkit/auto-animate"; +import { actionRunningPresentation } from "@t3tools/shared/actionResume"; import React, { useCallback, useEffect, @@ -905,12 +906,14 @@ export const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThr )} - {thread.actionResume?.outcome === "running" && threadStatus?.label !== "Waiting" ? ( + {thread.actionResume?.outcome === "running" && + actionRunningPresentation(thread.actionResume).state === "waiting" && + threadStatus?.label !== "Waiting" ? ( } @@ -920,7 +923,9 @@ export const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThr className="size-1.5 rounded-full bg-yellow-500 dark:bg-yellow-300" /> - Waiting for {thread.actionResume.actionName} + + {actionRunningPresentation(thread.actionResume).summary} + ) : null} {threadStatus && ( diff --git a/apps/web/src/components/Sidebar.logic.test.ts b/apps/web/src/components/Sidebar.logic.test.ts index d8456fd15899..ebd1a0ffba92 100644 --- a/apps/web/src/components/Sidebar.logic.test.ts +++ b/apps/web/src/components/Sidebar.logic.test.ts @@ -757,6 +757,20 @@ describe("resolveSidebarThreadStatus", () => { ).toBe("approval"); }); + it("reports Working when a running Action publishes working progress", () => { + const actionResume = { + outcome: "running", + actionName: "QA", + progress: { + version: 1, + state: "working", + summary: "Running checks", + updatedAt: "2026-08-29T12:00:00.000Z", + }, + } as never; + expect(resolveSidebarThreadStatus({ ...idle, session: null, actionResume })).toBe("working"); + }); + it("reports failed only while the session status is error", () => { expect( resolveSidebarThreadStatus({ diff --git a/apps/web/src/components/Sidebar.logic.ts b/apps/web/src/components/Sidebar.logic.ts index 07c2c2d4ac5d..518974d4dd42 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -2,6 +2,7 @@ import * as React from "react"; import { defaultAnimateLayoutChanges, type AnimateLayoutChanges } from "@dnd-kit/sortable"; import type { ContextMenuItem } from "@t3tools/contracts"; import type { SidebarProjectSortOrder, SidebarThreadSortOrder } from "@t3tools/contracts/settings"; +import { actionRunningPresentation } from "@t3tools/shared/actionResume"; import { activeThreadAnchorTimestampMs, getThreadSortTimestamp, @@ -518,7 +519,7 @@ export function resolveSidebarThreadStatus(thread: SidebarThreadStatusInput): Si return "monitoring"; } if (thread.actionResume?.outcome === "running") { - return "waiting"; + return actionRunningPresentation(thread.actionResume).state; } return "ready"; } @@ -812,10 +813,17 @@ export function resolveThreadStatusPill(input: { } if (thread.actionResume?.outcome === "running") { + const action = actionRunningPresentation(thread.actionResume); return { - label: "Waiting", - colorClass: "text-yellow-700 dark:text-yellow-300/90", - dotClass: "bg-yellow-500 dark:bg-yellow-300/90", + label: action.label, + colorClass: + action.state === "working" + ? "text-sky-600 dark:text-sky-300/80" + : "text-yellow-700 dark:text-yellow-300/90", + dotClass: + action.state === "working" + ? "bg-sky-500 dark:bg-sky-300/80" + : "bg-yellow-500 dark:bg-yellow-300/90", pulse: false, }; } diff --git a/apps/web/src/components/Sidebar.tsx b/apps/web/src/components/Sidebar.tsx index 9ec97cc1c412..144e1dd9fb33 100644 --- a/apps/web/src/components/Sidebar.tsx +++ b/apps/web/src/components/Sidebar.tsx @@ -32,6 +32,7 @@ import { } from "@t3tools/client-runtime/environment"; import type { ScopedThreadRef, ThreadId } from "@t3tools/contracts"; import type { EnvironmentIconColor, TimestampFormat } from "@t3tools/contracts/settings"; +import { actionRunningPresentation } from "@t3tools/shared/actionResume"; import { AlarmClockIcon, AlarmClockOffIcon, @@ -158,7 +159,6 @@ import { terminalStatusFromRunningIds, threadChangeRequestSnapshotsAtom, type ThreadChangeRequestSnapshot, - type TerminalStatusIndicator, useLinkedThreadPullRequest, } from "./ThreadStatusIndicators"; import { @@ -716,6 +716,8 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { const closeTerminal = useAtomCommand(terminalEnvironment.close, "cancel Project Action"); const ensureTerminal = useTerminalUiStateStore((state) => state.ensureTerminal); const actionResume = thread.actionResume ?? null; + const actionPresentation = + actionResume?.outcome === "running" ? actionRunningPresentation(actionResume) : null; const openActionTerminal = useCallback( (event: ReactMouseEvent) => { event.preventDefault(); @@ -769,6 +771,17 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { // switching sidebars must not light up every historical thread as unread. const isUnread = hasUnseenCompletion({ ...thread, lastVisitedAt }); const status = resolveSidebarThreadStatus(thread); + const actionIsPrimary = + actionPresentation !== null && + !isCleanupPending && + !isCleanupFailed && + !thread.hasPendingApprovals && + !thread.hasPendingUserInput && + thread.session?.status !== "running" && + thread.session?.status !== "starting" && + thread.session?.status !== "error" && + thread.backgroundLiveness !== "working" && + thread.backgroundLiveness !== "monitoring"; // A woken thread reappears at its original position (the sort is // deliberately static), so the pill has to carry the weight. Snoozing is // an explicit act, so the pill clears only when the user re-engages: @@ -1466,7 +1479,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { while the other controls appear beside it. */} {topStatus ? ( - status === "waiting" && actionResume !== null ? ( + actionIsPrimary && actionResume !== null && actionPresentation !== null ? ( event.stopPropagation()} className={cn( "inline-flex cursor-pointer items-center gap-1 rounded-sm font-medium outline-none hover:underline focus-visible:ring-2 focus-visible:ring-ring", @@ -1490,7 +1503,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { } > - Waiting + {actionPresentation.label}

{actionResume.actionName}

- Running for + {actionPresentation.summary} +

+

+ {actionPresentation.label} for{" "} +

@@ -1547,21 +1564,21 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: { topStatus.className, )} > - {actionResume?.outcome === "running" && status !== "waiting" ? ( + {actionPresentation?.state === "waiting" && + actionResume !== null && + !actionIsPrimary ? ( } > - - Waiting for {actionResume.actionName} - + {actionPresentation.summary} ) : null} {topStatus.icon === "working" || topStatus.icon === "cleanup" ? ( diff --git a/apps/web/src/components/chat/ComposerActionResume.test.tsx b/apps/web/src/components/chat/ComposerActionResume.test.tsx index 92647a8e7b18..89b0050bdc59 100644 --- a/apps/web/src/components/chat/ComposerActionResume.test.tsx +++ b/apps/web/src/components/chat/ComposerActionResume.test.tsx @@ -38,7 +38,6 @@ describe("ComposerActionResumeBadge", () => { expect(markup).toContain("text-warning-foreground"); expect(markup).toContain("lucide-rotate-ccw-clock"); expect(markup).toContain("Deploy preview"); - expect(markup).toContain("Running"); expect(markup).toContain('aria-expanded="false"'); }); diff --git a/apps/web/src/components/chat/ComposerActionResume.tsx b/apps/web/src/components/chat/ComposerActionResume.tsx index 58789f6d5327..13c921a531f0 100644 --- a/apps/web/src/components/chat/ComposerActionResume.tsx +++ b/apps/web/src/components/chat/ComposerActionResume.tsx @@ -1,4 +1,5 @@ import type { ActionResumeState } from "@t3tools/contracts"; +import { actionRunningPresentation } from "@t3tools/shared/actionResume"; import { TerminalIcon, XIcon } from "lucide-react"; import { memo, useEffect, useState } from "react"; @@ -42,7 +43,8 @@ export const ComposerActionResumeBadge = memo(function ComposerActionResumeBadge readonly onToggle: () => void; readonly placement?: "inline" | "tab"; }) { - const label = `${action.action.actionName} is running and will resume the agent when the thread is idle`; + const presentation = actionRunningPresentation(action.action); + const label = `${action.action.actionName}: ${presentation.label}. ${presentation.summary}. The agent will resume when the thread is idle.`; if (placement === "inline") { return ( @@ -52,12 +54,16 @@ export const ComposerActionResumeBadge = memo(function ComposerActionResumeBadge aria-expanded={expanded} aria-label={label} disabled={disabled} - className="shrink-0 gap-1 px-1.5 text-warning-foreground" + className={cn( + "shrink-0 gap-1 px-1.5", + presentation.state === "working" ? "text-info-foreground" : "text-warning-foreground", + )} onClick={onToggle} onPointerDown={(event) => event.preventDefault()} > - {action.action.actionName} + {presentation.summary} + {presentation.label} ); @@ -65,9 +71,12 @@ export const ComposerActionResumeBadge = memo(function ComposerActionResumeBadge return (
@@ -101,11 +110,12 @@ export const ComposerActionResumeDrawer = memo(function ComposerActionResumeDraw readonly onCollapse: () => void; readonly onOpenTerminal: () => void; }) { + const presentation = actionRunningPresentation(action.action); return (
+
+
+ Current status +
+

{presentation.summary}

+
Command diff --git a/apps/web/src/components/chat/MessagesTimeline.test.tsx b/apps/web/src/components/chat/MessagesTimeline.test.tsx index 3fb0650cb64d..0bdccab23133 100644 --- a/apps/web/src/components/chat/MessagesTimeline.test.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.test.tsx @@ -244,6 +244,7 @@ describe("MessagesTimeline", () => { actionId: "run-full-ci", runId: "run-full-ci-1", validatedStatus: "succeeded", + lifecycleOutcome: "succeeded", exitCode: 0, report: undefined, output: "full output hidden while collapsed\n[lastcode:ci] Summary: all checks passed", @@ -256,13 +257,11 @@ describe("MessagesTimeline", () => { />, ); - expect(markup).toContain("Action completed: Run Full CI Status: 0"); + expect(markup).toContain("Succeeded: Run Full CI"); expect(markup).toContain("[lastcode:ci] Summary: all checks passed"); expect(markup).not.toContain("full output hidden while collapsed"); expect(markup).not.toContain('aria-expanded="false"'); expect(markup).toContain("Detailed output retained in the Action terminal."); - expect(markup).toContain("border-warning/28 bg-warning/8"); - expect(markup).toContain("text-warning-foreground"); }); it("shows the validated outcome when an Action has no exit code", () => { @@ -271,6 +270,7 @@ describe("MessagesTimeline", () => { actionId: "wait-for-pr", runId: "wait-for-pr-1", validatedStatus: "was cancelled by the user", + lifecycleOutcome: "cancelled_by_user", exitCode: null, report: undefined, output: "Cancellation requested.", @@ -283,8 +283,7 @@ describe("MessagesTimeline", () => { />, ); - expect(markup).toContain("Action completed: Wait for PR Status: was cancelled by the user"); - expect(markup).not.toContain("Status: unavailable"); + expect(markup).toContain("Cancelled: Wait for PR"); }); it("renders a feedback command and its pending response as normal thread messages", () => { diff --git a/apps/web/src/components/chat/MessagesTimeline.tsx b/apps/web/src/components/chat/MessagesTimeline.tsx index c168e7ebc5f2..8b9908af56df 100644 --- a/apps/web/src/components/chat/MessagesTimeline.tsx +++ b/apps/web/src/components/chat/MessagesTimeline.tsx @@ -18,7 +18,7 @@ const EMPTY_AGENT_PANEL_MODEL = emptyAgentPanelModel(); const NOOP_OPEN_AGENTS = () => {}; const NOOP_DOWNLOAD_ATTACHMENT = (_attachment: ChatFileAttachment) => {}; import { resolveChatListAnchoredEndSpace } from "@t3tools/shared/chatList"; -import { parseActionResumeFollowUp } from "@t3tools/shared/actionResume"; +import { actionResultPresentation, parseActionResumeFollowUp } from "@t3tools/shared/actionResume"; import { createContext, Fragment, @@ -1470,26 +1470,42 @@ function SystemTimelineRow({ row }: { row: Extract - + - Action completed: {actionFollowUp.actionName} Status: {status} + {presentation.label}: {actionFollowUp.actionName} ); return ( -
+
{actionFollowUp.detailedOutputAvailable ? ( -
+
{heading}
) : (