diff --git a/apps/mobile/src/connection/runtime.ts b/apps/mobile/src/connection/runtime.ts index 58aeb64992bd3..9976dfe60f7d0 100644 --- a/apps/mobile/src/connection/runtime.ts +++ b/apps/mobile/src/connection/runtime.ts @@ -39,6 +39,7 @@ const layerProvidedClientConnection = layerSnapshotLoader.pipe( Connection.layerWithOptions({ usageLimitSources: true, memoryPressure: true, + quietRuns: true, usageLimitsCommand: true, }), ), diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index 6695e7963eff0..85f672e52e476 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -160,7 +160,11 @@ import { useAtomCommand } from "../../state/use-atom-command"; import { orchestrationEnvironment } from "../../state/orchestration"; import { useDelayedStatus } from "../../lib/useDelayedStatus"; import type { PendingUserInputDraftAnswer } from "@t3tools/client-runtime/state/pending-user-input"; -import type { RunStopPresentation } from "@t3tools/client-runtime/state/thread-execution"; +import { + quietRunId, + quietRunSince, + type RunStopPresentation, +} from "@t3tools/client-runtime/state/thread-execution"; import type { PendingApproval, PendingUserInput, @@ -550,6 +554,12 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { const canForceStop = useAtomValue(orchestrationEnvironment.forceStopThread.permissionAtom(props.environmentId)) && props.serverConfig?.environment.capabilities.threadForceStop === true; + // The run the server saw go quiet, not the latest one: a queued follow-up can be newer. + const quietRunTarget = quietRunId(props.serverConfig, props.selectedThread) ?? undefined; + const quietSince = + props.activeWorkStartedAt === null + ? null + : quietRunSince(props.serverConfig, props.selectedThread); const [forceStoppingKey, setForceStoppingKey] = useState(null); // The run that ignored Stop, so a confirmation left open cannot hit a newer run. const forceStop = async (runId: RunId | undefined) => { @@ -580,6 +590,26 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { }; const [statusRequest, setStatusRequest] = useState<{ threadKey: string } | null>(null); const statusPending = statusRequest?.threadKey === selectedThreadKey; + const stopQuietRun = async () => { + if (statusPending || !props.canOperateThread) return; + const request = { threadKey: selectedThreadKey }; + setStatusRequest(request); + const result = await interruptThreadTurn({ + environmentId: props.environmentId, + input: + quietRunTarget === undefined + ? { threadId: props.selectedThread.id } + : { threadId: props.selectedThread.id, runId: quietRunTarget }, + }); + setStatusRequest((current) => (current === request ? null : current)); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + Dialog.alert( + "Could not stop the turn", + error instanceof Error ? error.message : "Try again.", + ); + } + }; const [lifecycleNow, setLifecycleNow] = useState(() => new Date().toISOString()); useFocusEffect( useCallback(() => { @@ -723,6 +753,46 @@ function ThreadDetailBody(props: ThreadDetailScreenProps) { }, }; } + if (quietSince !== null && contentPresentationKind === "ready") { + return { + kind: "quiet", + quietSince, + action: { + label: + forceStoppingKey === selectedThreadKey + ? "Force stopping..." + : statusPending + ? "Stopping..." + : "Stop", + accessibilityLabel: "Stop the quiet turn", + disabled: + !props.canOperateThread || statusPending || forceStoppingKey === selectedThreadKey, + onPress: () => + Dialog.alert( + "Stop this turn?", + canForceStop + ? "The agent has sent nothing for a while. A long tool may still be running. Force stop ends the turn and kills the agent's processes, unless other working threads share them." + : "The agent has sent nothing for a while. A long tool may still be running.", + [ + { text: "Cancel", style: "cancel" }, + { + text: "Stop", + onPress: () => void stopQuietRun(), + }, + ...(canForceStop + ? [ + { + text: "Force stop", + style: "destructive" as const, + onPress: () => void forceStop(quietRunTarget), + }, + ] + : []), + ], + ), + }, + }; + } if (props.isCompacting && contentPresentationKind === "ready") { return { kind: "compacting" }; } diff --git a/apps/mobile/src/features/threads/floating-working-control.tsx b/apps/mobile/src/features/threads/floating-working-control.tsx index c1440535e9cad..252bb6da44314 100644 --- a/apps/mobile/src/features/threads/floating-working-control.tsx +++ b/apps/mobile/src/features/threads/floating-working-control.tsx @@ -85,7 +85,8 @@ export function FloatingWorkingControl(props: { const action = props.status?.kind === "background" || props.status?.kind === "lifecycle" || - props.status?.kind === "stop" + props.status?.kind === "stop" || + props.status?.kind === "quiet" ? props.status.action : null; const [agentsWidth, setAgentsWidth] = useState(0); @@ -179,7 +180,9 @@ export function FloatingWorkingControl(props: { > ); } + if (props.status.kind === "quiet") { + return ( + + + + + ); + } if (props.status.kind === "stop") { return ( - Working {formatWorkingDuration(props.startedAt, nowMs)} + {props.prefix ?? "Working"} {formatWorkingDuration(props.startedAt, nowMs)} ); } diff --git a/apps/mobile/src/features/threads/floating-working-status.ts b/apps/mobile/src/features/threads/floating-working-status.ts index 2892b4aafa51b..febb3de9bf772 100644 --- a/apps/mobile/src/features/threads/floating-working-status.ts +++ b/apps/mobile/src/features/threads/floating-working-status.ts @@ -31,6 +31,12 @@ export type FloatingWorkingStatus = readonly label: string; readonly action: FloatingStatusAction; } + | { + readonly kind: "quiet"; + /** When the provider last sent anything for the running turn. */ + readonly quietSince: string; + readonly action: FloatingStatusAction; + } | { readonly kind: "stop"; readonly label: string; 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 5b0c1fd2f5284..a45322e9c52cb 100644 --- a/apps/mobile/src/features/threads/thread-list-v2-items.tsx +++ b/apps/mobile/src/features/threads/thread-list-v2-items.tsx @@ -13,6 +13,9 @@ import { CustomSnoozeSheet } from "./CustomSnoozeSheet"; import { appAtomRegistry } from "../../state/atom-registry"; import { threadArrangementOpenAtom } from "../../state/thread-order"; import { threadEnvironment } from "../../state/threads"; +import { serverEnvironment } from "../../state/server"; +import { useAtomValue } from "@effect/atom-react"; +import { quietRunSince } from "@t3tools/client-runtime/state/thread-execution"; import { useAtomCommand } from "../../state/use-atom-command"; import { useVisibleSecondClock } from "./use-visible-second-clock"; import type { ThreadMoveDestination } from "./threadOrder"; @@ -132,6 +135,12 @@ const STATUS_LABEL_BY_STATUS: Partial> = iconTintClassName: "accent-warning-foreground", }, }; +const QUIET_STATUS_LABEL: StatusLabel = { + label: "No activity", + icon: "timer", + className: "text-warning-foreground", + iconTintClassName: "accent-warning-foreground", +}; const DONE_STATUS_LABEL: StatusLabel = { label: "Done", icon: "checkmark.circle", @@ -688,10 +697,24 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { const status = resolveThreadListV2Status(thread); const isUnread = status === "ready" && threadHasUnseenCompletion(thread); - const workingLabel = STATUS_LABEL_BY_STATUS[status]; + const quietSince = useAtomValue( + serverEnvironment.configValueAtom(thread.environmentId), + (config) => (status === "working" ? quietRunSince(config, thread) : null), + ); + const workingLabel = quietSince !== null ? QUIET_STATUS_LABEL : STATUS_LABEL_BY_STATUS[status]; + const rowAccessibilityLabel = [ + thread.title, + quietSince !== null ? QUIET_STATUS_LABEL.label : null, + props.hasQueuedMessages ? "messages queued to send" : null, + ] + .filter((part) => part !== null) + .join(", "); const showsWoke = status === "ready" && props.wokeAt !== null; const statusLabel = - (status === "working" && workingLabel !== undefined && thread.goal?.status === "active" + (status === "working" && + quietSince === null && + workingLabel !== undefined && + thread.goal?.status === "active" ? { ...workingLabel, label: "Goal" } : workingLabel) ?? (isUnread ? DONE_STATUS_LABEL : undefined); const recede = shouldRecedeThreadRow({ status, selected }); @@ -1145,7 +1168,9 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { > {statusLabel.label} {status === "working" ? ( - + ) : status === "waiting" ? ( ) : null} @@ -1299,9 +1324,7 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { interactionOpacity={rowAppearance.interactionOpacity} className={rowAppearance.className} accessibilityHint={swipeAccessibilityHint} - accessibilityLabel={ - props.hasQueuedMessages ? `${thread.title}, messages queued to send` : thread.title - } + accessibilityLabel={rowAccessibilityLabel} accessibilityRole="button" accessibilityState={{ selected }} accessibilityActions={rowAccessibilityActions} @@ -1332,9 +1355,7 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { interactionClassName={rowAppearance.interactionClassName} interactionOpacity={rowAppearance.interactionOpacity} accessibilityHint={swipeAccessibilityHint} - accessibilityLabel={ - props.hasQueuedMessages ? `${thread.title}, messages queued to send` : thread.title - } + accessibilityLabel={rowAccessibilityLabel} accessibilityRole="button" accessibilityState={{ selected }} className={rowAppearance.className} diff --git a/apps/mobile/src/state/server.ts b/apps/mobile/src/state/server.ts index 6e319b276f74e..625fae5d10b53 100644 --- a/apps/mobile/src/state/server.ts +++ b/apps/mobile/src/state/server.ts @@ -13,6 +13,7 @@ export const serverEnvironment = createServerEnvironmentAtoms(connectionAtomRunt initialConfigValueAtom: environmentSession.initialConfigValueAtom, usageLimitSources: true, memoryPressure: true, + quietRuns: true, usageLimitsCommand: true, }); export const describeOutdatedServer = createOutdatedServerDescriptionCommand(connectionAtomRuntime); diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index fe9c092526672..4164bc44da791 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -242,6 +242,7 @@ export const make = Effect.gen(function* () { threadSnooze: true, threadForceStop: true, memoryPressure: true, + quietRuns: true, environmentThemes: true, usageLimitSources: true, providerWorkspaceSnapshotsOnDemand: true, diff --git a/apps/server/src/orchestration-v2/EffectWorker.test.ts b/apps/server/src/orchestration-v2/EffectWorker.test.ts index 2329f2aa20cf0..6f3f1f3f6c65b 100644 --- a/apps/server/src/orchestration-v2/EffectWorker.test.ts +++ b/apps/server/src/orchestration-v2/EffectWorker.test.ts @@ -21,6 +21,7 @@ import * as Exit from "effect/Exit"; import * as Layer from "effect/Layer"; import * as Option from "effect/Option"; import * as Ref from "effect/Ref"; +import * as Stream from "effect/Stream"; import * as CheckpointRollbackService from "./CheckpointRollbackService.ts"; import * as CheckpointService from "./CheckpointService.ts"; @@ -113,6 +114,7 @@ function layerExecutorFor(input: { detach: () => record("detach"), otherWorkingThreads: () => Effect.succeed([]), releaseUnlessBusy: () => Effect.succeed(true), + quietRuns: Stream.empty, }), ), Layer.succeed( diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts index 5bbb3926e2b66..b6e22c52b068e 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.test.ts @@ -10,6 +10,7 @@ import { type OrchestrationV2ProviderCapabilities, type OrchestrationV2ProviderSession, type OrchestrationV2ProviderThread, + PlanId, type Project, ProjectId, ProviderDriverKind, @@ -433,6 +434,7 @@ const makeProviderAdapter = Effect.fnUntraced(function* ( resumeCount: current.resumeCount + 1, })).pipe(Effect.as(threadInput.providerThread)), startTurn: () => options.startTurn ?? Effect.void, + compactThread: () => Effect.void, steerTurn: () => Effect.void, interruptTurn: () => Ref.update(state, (current) => ({ @@ -1409,6 +1411,223 @@ it.effect("ProviderSessionManagerV2 gives up on a provider that never confirms a }), ); +it.effect.each(["startTurn", "compactThread"] as const)( + "ProviderSessionManagerV2 flags a quiet %s, despite an older run's late terminal, until the provider speaks", + (start) => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const threadId = ThreadId.make("thread-provider-session-manager-quiet-run"); + const providerSessionId = yield* idAllocator.allocate.providerSession({ + providerInstanceId: modelSelection.instanceId, + threadId, + }); + yield* eventSink.write({ + events: [yield* makeThreadCreatedEvent({ idAllocator, threadId, now })], + }); + const runtime = yield* manager.open({ + threadId, + providerSessionId, + modelSelection, + runtimePolicy, + }); + const providerThread = makeProviderThread({ + idAllocator, + threadId, + providerSessionId, + now, + }); + const quiet = yield* manager.quietRuns.pipe( + Stream.map((runs) => runs.map((run) => [run.threadId, run.runId])), + Stream.take(3), + Stream.runCollect, + Effect.forkChild, + ); + const startRun = Effect.fnUntraced(function* (ordinal: number) { + const runId = idAllocator.derive.run({ threadId, ordinal }); + yield* runtime[start]!({ + appThread: (yield* projectionStore.getThreadProjection(threadId)).thread, + threadId, + runId, + runOrdinal: ordinal, + providerTurnOrdinal: ordinal, + attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }), + rootNodeId: idAllocator.derive.rootNode({ runId }), + providerThread, + message: { + createdBy: "user", + creationSource: "web", + messageId: yield* idAllocator.allocate.message({ threadId, ordinal }), + text: "turn", + attachments: [], + }, + modelSelection, + runtimePolicy, + }); + }); + const publish = Effect.fnUntraced(function* ( + event: ProviderAdapter.ProviderAdapterV2Event, + ) { + // Each read ends its subscription, so every event needs its own. + const subscription = yield* runtime.subscribeEvents!; + yield* Queue.offer( + (yield* Ref.get(state)).eventQueues.get(String(providerSessionId))!, + event, + ); + // Observing the event ensures the manager handled it first. + yield* subscription.events.pipe(Stream.runHead); + }); + yield* startRun(1); + yield* startRun(2); + // Late enough that refreshing run 2 would push its flag past the threshold. + yield* TestClock.adjust(ProviderSessionManager.QUIET_RUN_AFTER_MS - 60_000); + yield* publish({ + type: "turn.terminal", + driver: CODEX_DRIVER, + providerThreadId: providerThread.id, + providerTurnId: idAllocator.derive.providerTurn({ + driver: CODEX_DRIVER, + nativeTurnId: "native-turn-stopped-run", + }), + runOrdinal: 1, + status: "interrupted", + failure: null, + threadDisposition: "reusable", + }); + yield* TestClock.adjust(60_000); + + yield* publish({ type: "provider_thread.updated", driver: CODEX_DRIVER, providerThread }); + yield* TestClock.adjust("30 seconds"); + assert.deepEqual(yield* Fiber.join(quiet), [ + [], + [[threadId, idAllocator.derive.run({ threadId, ordinal: 2 })]], + [], + ]); + }).pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 60 * 60_000 }))); + }), +); + +it.effect( + "ProviderSessionManagerV2 keeps a quiet thread flagged while a sibling in its shared session works", + () => + Effect.gen(function* () { + const state = yield* Ref.make(emptyState); + yield* Effect.gen(function* () { + const eventSink = yield* EventSink.EventSinkV2; + const idAllocator = yield* IdAllocator.IdAllocatorV2; + const manager = yield* ProviderSessionManager.ProviderSessionManagerV2; + const projectionStore = yield* ProjectionStore.ProjectionStoreV2; + const now = yield* DateTime.now; + const quietThread = ThreadId.make("thread-provider-session-manager-quiet-sibling-a"); + const busyThread = ThreadId.make("thread-provider-session-manager-quiet-sibling-b"); + const providerSessionId = idAllocator.derive.providerSession({ + providerInstanceId: modelSelection.instanceId, + }); + yield* eventSink.write({ + events: [ + yield* makeThreadCreatedEvent({ idAllocator, threadId: quietThread, now }), + yield* makeThreadCreatedEvent({ idAllocator, threadId: busyThread, now }), + ], + }); + const runtime = yield* manager.open({ + threadId: quietThread, + providerSessionId, + modelSelection, + runtimePolicy, + }); + yield* manager.open({ + threadId: busyThread, + providerSessionId, + modelSelection, + runtimePolicy, + }); + const providerThreadOf = (threadId: ThreadId) => + makeProviderThread({ + idAllocator, + threadId, + providerSessionId, + now, + nativeThreadId: `native-${threadId}`, + }); + const quiet = yield* manager.quietRuns.pipe( + Stream.map((runs) => runs.map((run) => run.threadId)), + Stream.take(4), + Stream.runCollect, + Effect.forkChild, + ); + for (const threadId of [quietThread, busyThread]) { + const runId = idAllocator.derive.run({ threadId, ordinal: 1 }); + yield* runtime.startTurn({ + appThread: (yield* projectionStore.getThreadProjection(threadId)).thread, + threadId, + runId, + runOrdinal: 1, + providerTurnOrdinal: 1, + attemptId: idAllocator.derive.runAttempt({ runId, attemptOrdinal: 1 }), + rootNodeId: idAllocator.derive.rootNode({ runId }), + providerThread: providerThreadOf(threadId), + message: { + createdBy: "user", + creationSource: "web", + messageId: yield* idAllocator.allocate.message({ threadId, ordinal: 1 }), + text: "turn", + attachments: [], + }, + modelSelection, + runtimePolicy, + }); + } + yield* TestClock.adjust(ProviderSessionManager.QUIET_RUN_AFTER_MS); + + const busyRunId = idAllocator.derive.run({ threadId: busyThread, ordinal: 1 }); + for (const event of [ + { + type: "provider_thread.updated", + driver: CODEX_DRIVER, + providerThread: providerThreadOf(busyThread), + }, + // Session-level events that still name the sibling they belong to. + { + type: "plan.updated", + driver: CODEX_DRIVER, + plan: { + id: PlanId.make("plan-quiet-sibling"), + threadId: busyThread, + runId: busyRunId, + nodeId: idAllocator.derive.rootNode({ runId: busyRunId }), + status: "active", + kind: "todo_list", + steps: [], + }, + }, + ] satisfies ReadonlyArray) { + // Each read ends its subscription, so every event needs its own. + const subscription = yield* runtime.subscribeEvents!; + yield* Queue.offer( + (yield* Ref.get(state)).eventQueues.get(String(providerSessionId))!, + event, + ); + yield* subscription.events.pipe(Stream.runHead); + } + yield* TestClock.adjust("30 seconds"); + // Force stop detaches a thread from a shared session; no terminal ever arrives for it. + yield* manager.detach({ providerSessionId, threadId: quietThread }); + yield* TestClock.adjust("30 seconds"); + assert.deepEqual(yield* Fiber.join(quiet), [ + [], + [quietThread, busyThread].toSorted(), + [quietThread], + [], + ]); + }).pipe(Effect.provide(layerTest({ state, idleTimeoutMs: 60 * 60_000 }))); + }), +); + it.effect( "ProviderSessionManagerV2 keeps a shared session a sibling is busy on when releasing", () => diff --git a/apps/server/src/orchestration-v2/ProviderSessionManager.ts b/apps/server/src/orchestration-v2/ProviderSessionManager.ts index 03a7278774a64..80ce2de9d82ba 100644 --- a/apps/server/src/orchestration-v2/ProviderSessionManager.ts +++ b/apps/server/src/orchestration-v2/ProviderSessionManager.ts @@ -10,6 +10,8 @@ import { OrchestrationV2RuntimeRequest, ProviderInstanceId, ProviderSessionId, + type RunId, + type ServerQuietRun, ThreadId, type ProviderThreadId, } from "@t3tools/contracts"; @@ -59,6 +61,9 @@ const DEFAULT_MAX_IDLE_PIN_MS = 4 * 60 * 60 * 1000; const RELEASE_SCOPE_CLOSE_TIMEOUT_MS = 30 * 1000; /** Past adapters' own 10 s stop fallbacks, so they settle first when the provider answers at all. */ export const PROVIDER_INTERRUPT_TIMEOUT_MS = 15 * 1000; +/** Tools legitimately run long, so this flags a quiet turn for the user, never stops it. */ +export const QUIET_RUN_AFTER_MS = 10 * 60 * 1000; +const QUIET_RUN_CHECK_INTERVAL = "30 seconds"; const busyTurnPrefix = (providerThreadId: ProviderThreadId) => `${providerThreadId}#`; const busyTurnKey = (providerThreadId: ProviderThreadId, runOrdinal: number) => @@ -151,6 +156,45 @@ export const ProviderSessionManagerV2Error = Schema.Union([ ]); export type ProviderSessionManagerV2Error = typeof ProviderSessionManagerV2Error.Type; +interface ActivityTarget { + readonly providerThreadId?: ProviderThreadId; + readonly threadId?: ThreadId; + /** Set when the event belongs to one run; an older run's late event must not count. */ + readonly runOrdinal?: number; +} + +/** The turn an event speaks for, so a busy sibling in a shared session cannot hide a quiet one. */ +function eventActivityTarget( + event: ProviderAdapter.ProviderAdapterV2Event, +): ActivityTarget | undefined { + const onProviderThread = (providerThreadId: ProviderThreadId | null | undefined) => + providerThreadId == null ? undefined : { providerThreadId }; + switch (event.type) { + case "turn_item.updated": + return onProviderThread(event.turnItem.providerThreadId); + case "provider_thread.updated": + return { providerThreadId: event.providerThread.id }; + case "provider_turn.updated": + return { providerThreadId: event.providerTurn.providerThreadId }; + case "node.updated": + return onProviderThread(event.node.providerThreadId); + case "subagent.updated": + return onProviderThread(event.subagent.providerThreadId); + case "turn.terminal": + return { providerThreadId: event.providerThreadId, runOrdinal: event.runOrdinal }; + case "message.updated": + return { threadId: event.message.threadId }; + case "command_output.delta": + return { threadId: event.threadId }; + case "plan.updated": + return { threadId: event.plan.threadId }; + case "runtime_request.updated": + return event.threadId === undefined ? undefined : { threadId: event.threadId }; + default: + return undefined; + } +} + export interface ProviderSessionManagerV2Shape { readonly shutdown: Effect.Effect; readonly open: (input: { @@ -200,6 +244,11 @@ export interface ProviderSessionManagerV2Shape { readonly reason: ProviderSessionReleaseReason; readonly detail?: string; }) => Effect.Effect; + /** + * Threads whose running turn has had no provider event for `QUIET_RUN_AFTER_MS`, sorted by + * thread. Changes only; checks while someone listens. + */ + readonly quietRuns: Stream.Stream>; /** * Other threads with a run or background work on this session's shared process; empty when * it serves one thread. @@ -402,6 +451,41 @@ export const layerWithOptions = ( }), ); const sessions = yield* Ref.make(new Map()); + // Running turns by provider thread, and when the provider last sent anything for them. + const runActivity = new Map< + ProviderThreadId, + { + readonly threadId: ThreadId; + readonly providerSessionId: ProviderSessionId; + readonly runId: RunId; + readonly runOrdinal: number; + lastEventMs: number; + } + >(); + const touchRunActivity = ( + providerSessionId: ProviderSessionId, + target: ActivityTarget | undefined, + ) => + Effect.map(Clock.currentTimeMillis, (nowMs) => { + const tracked = + target?.providerThreadId === undefined + ? undefined + : runActivity.get(target.providerThreadId); + if (tracked !== undefined) { + if (target?.runOrdinal === undefined || target.runOrdinal === tracked.runOrdinal) + tracked.lastEventMs = nowMs; + return; + } + // Session events and native subagents' own threads count for every turn in the + // session; an event naming its app thread counts only for that thread's turns. + for (const run of runActivity.values()) { + if ( + run.providerSessionId === providerSessionId && + (target?.threadId === undefined || run.threadId === target.threadId) + ) + run.lastEventMs = nowMs; + } + }); // Retry per released entry; reusing ids must not lose cleanup for earlier threads. const releaseRecordRetries = yield* FiberSet.make(); const nextSubscriberId = yield* Ref.make(0); @@ -1019,6 +1103,10 @@ export const layerWithOptions = ( onNone: () => Effect.void, onSome: (entry) => Effect.gen(function* () { + for (const [providerThreadId, run] of runActivity) { + if (run.providerSessionId === input.providerSessionId) + runActivity.delete(providerThreadId); + } const releasedBackgroundTasks = yield* readReleasedBackgroundTasks({ entry, reason: input.reason, @@ -1463,6 +1551,9 @@ export const layerWithOptions = ( if (entry === undefined) { return current; } + // A stopped run's late terminal must not clear the newer run on this thread. + if (runActivity.get(providerThreadId)?.runOrdinal === runOrdinal) + runActivity.delete(providerThreadId); const busyTurns = new Set(entry.busyTurns); busyTurns.delete(busyTurnKey(providerThreadId, runOrdinal)); const updated = new Map(current); @@ -1697,6 +1788,7 @@ export const layerWithOptions = ( >, ): ProviderAdapter.ProviderAdapterV2SessionRuntime => { const providerSessionId = runtime.providerSessionId; + const compactThread = runtime.compactThread; const subscribeEvents = makeEventSubscription(eventSubscribers); const turnMetrics = (operation: string, model?: string) => withMetrics({ @@ -1857,6 +1949,16 @@ export const layerWithOptions = ( }), ), ), + Effect.andThen(Clock.currentTimeMillis), + Effect.map((nowMs) => { + runActivity.set(input.providerThread.id, { + threadId: input.threadId, + providerSessionId, + runId: input.runId, + runOrdinal: input.runOrdinal, + lastEventMs: nowMs, + }); + }), ), () => runtime.startTurn(input).pipe(turnMetrics("send", input.modelSelection.model)), @@ -1870,6 +1972,27 @@ export const layerWithOptions = ( ), ), ), + // A compaction is a provider turn too, so a frozen one must be flagged quiet. + ...(compactThread === undefined + ? {} + : { + compactThread: (input: ProviderAdapter.ProviderAdapterV2TurnInput) => + Clock.currentTimeMillis.pipe( + Effect.map((nowMs) => { + runActivity.set(input.providerThread.id, { + threadId: input.threadId, + providerSessionId, + runId: input.runId, + runOrdinal: input.runOrdinal, + lastEventMs: nowMs, + }); + }), + Effect.andThen(compactThread(input)), + Effect.onError(() => + Effect.sync(() => runActivity.delete(input.providerThread.id)), + ), + ), + }), steerTurn: (input) => observeActivity(providerSessionId, touchActivity(providerSessionId)).pipe( Effect.andThen(runtime.steerTurn(input).pipe(turnMetrics("steer"))), @@ -1945,9 +2068,13 @@ export const layerWithOptions = ( return entry.runtime.events.pipe( Stream.runForEach((event) => { if (shutdownSignal.received) return Effect.void; + const touched = touchRunActivity( + entry.runtime.providerSessionId, + eventActivityTarget(event), + ); // Live output bypasses persistence and subscribers so floods cannot delay domain events. if (event.type === "command_output.delta") { - return commandOutput.append(event); + return touched.pipe(Effect.andThen(commandOutput.append(event))); } const releaseLiveOutput = event.type === "turn_item.updated" && @@ -1964,76 +2091,81 @@ export const layerWithOptions = ( ) { stoppedByProvider = true; } - return observeActivity( - entry.runtime.providerSessionId, - event.type === "turn.terminal" - ? markIdle( - entry.runtime.providerSessionId, - event.providerThreadId, - event.runOrdinal, - ) - : touchActivity(entry.runtime.providerSessionId, { providerActivity: true }), - ).pipe( + return touched.pipe( Effect.andThen( - event.type === "provider_session.updated" - ? persistProviderSessionUpdate(entry, event) - : Effect.void, - ), - Effect.andThen( - Effect.gen(function* () { - if ( - event.type === "provider_thread.updated" && - event.providerThread.status === "error" - ) { - const threadId = event.providerThread.appThreadId; - yield* Ref.update(sessions, (sessions) => { - const key = sessionKey(entry.runtime.providerSessionId); - const latest = sessions.get(key); - if (latest?.runtime !== entry.runtime || threadId === null) return sessions; - const loadedProviderThreadKeyByThread = new Map( - latest.loadedProviderThreadKeyByThread, - ); - loadedProviderThreadKeyByThread.delete(threadId); - return new Map(sessions).set(key, { - ...latest, - loadedProviderThreadKeyByThread, - }); - }); - } - // Some providers can block before a run subscriber exists - // (project trust, login, or session-switch hooks). Persist - // their runless request artifacts directly so the normal T3 - // request UI can answer them and unblock session setup. - const threadId = sessionScopedRuntimeRequestThreadId(event); - if (threadId !== undefined) { - yield* Effect.gen(function* () { - const current = (yield* Ref.get(sessions)).get( - sessionKey(entry.runtime.providerSessionId), - ); - if (current?.runtime !== entry.runtime) return; - yield* providerEventIngestor - .ingestNormalized({ - providerSessionId: entry.runtime.providerSessionId, - providerInstanceId: entry.runtime.instanceId, - threadId, - event, - }) - .pipe( - Effect.mapError( - (cause) => - new ProviderAdapter.ProviderAdapterEventStreamError({ - driver: entry.runtime.driver, - providerSessionId: entry.runtime.providerSessionId, - cause, - }), - ), - ); - }).pipe(entry.requestEventPermit.withPermits(1)); - return; - } - yield* publishToSubscribers(entry.eventSubscribers, { type: "event", event }); - yield* releaseLiveOutput; - }), + observeActivity( + entry.runtime.providerSessionId, + event.type === "turn.terminal" + ? markIdle( + entry.runtime.providerSessionId, + event.providerThreadId, + event.runOrdinal, + ) + : touchActivity(entry.runtime.providerSessionId, { providerActivity: true }), + ).pipe( + Effect.andThen( + event.type === "provider_session.updated" + ? persistProviderSessionUpdate(entry, event) + : Effect.void, + ), + Effect.andThen( + Effect.gen(function* () { + if ( + event.type === "provider_thread.updated" && + event.providerThread.status === "error" + ) { + const threadId = event.providerThread.appThreadId; + yield* Ref.update(sessions, (sessions) => { + const key = sessionKey(entry.runtime.providerSessionId); + const latest = sessions.get(key); + if (latest?.runtime !== entry.runtime || threadId === null) + return sessions; + const loadedProviderThreadKeyByThread = new Map( + latest.loadedProviderThreadKeyByThread, + ); + loadedProviderThreadKeyByThread.delete(threadId); + return new Map(sessions).set(key, { + ...latest, + loadedProviderThreadKeyByThread, + }); + }); + } + // Some providers can block before a run subscriber exists + // (project trust, login, or session-switch hooks). Persist + // their runless request artifacts directly so the normal T3 + // request UI can answer them and unblock session setup. + const threadId = sessionScopedRuntimeRequestThreadId(event); + if (threadId !== undefined) { + yield* Effect.gen(function* () { + const current = (yield* Ref.get(sessions)).get( + sessionKey(entry.runtime.providerSessionId), + ); + if (current?.runtime !== entry.runtime) return; + yield* providerEventIngestor + .ingestNormalized({ + providerSessionId: entry.runtime.providerSessionId, + providerInstanceId: entry.runtime.instanceId, + threadId, + event, + }) + .pipe( + Effect.mapError( + (cause) => + new ProviderAdapter.ProviderAdapterEventStreamError({ + driver: entry.runtime.driver, + providerSessionId: entry.runtime.providerSessionId, + cause, + }), + ), + ); + }).pipe(entry.requestEventPermit.withPermits(1)); + return; + } + yield* publishToSubscribers(entry.eventSubscribers, { type: "event", event }); + yield* releaseLiveOutput; + }), + ), + ), ), ); }), @@ -2115,6 +2247,39 @@ export const layerWithOptions = ( ), ); + const quietRuns = yield* Stream.fromEffectSchedule( + Effect.map(Clock.currentTimeMillis, (nowMs) => { + // A thread is quiet only when none of its provider threads heard anything. + const latestByThread = new Map(); + for (const run of runActivity.values()) { + const latest = latestByThread.get(run.threadId); + if (latest === undefined || run.lastEventMs > latest.lastEventMs) + latestByThread.set(run.threadId, { lastEventMs: run.lastEventMs, runId: run.runId }); + } + return [...latestByThread] + .filter(([, latest]) => nowMs - latest.lastEventMs >= QUIET_RUN_AFTER_MS) + .map(([threadId, latest]) => ({ + threadId, + runId: latest.runId, + quietSince: DateTime.makeUnsafe(latest.lastEventMs), + })) + .toSorted((left, right) => left.threadId.localeCompare(right.threadId)); + }), + Schedule.spaced(QUIET_RUN_CHECK_INTERVAL), + ).pipe( + Stream.changesWith( + (previous, next) => + previous.length === next.length && + previous.every( + (run, index) => + run.threadId === next[index]?.threadId && + DateTime.toEpochMillis(run.quietSince) === + DateTime.toEpochMillis(next[index].quietSince), + ), + ), + Stream.share({ capacity: 1, strategy: "sliding", replay: 1 }), + ); + return ProviderSessionManagerV2.of({ shutdown, open: (input) => @@ -2371,6 +2536,7 @@ export const layerWithOptions = ( // Identity, not presence: open can install a replacement under the same id meanwhile. return before !== undefined && (yield* Ref.get(sessions)).get(key) !== before; }), + quietRuns, otherWorkingThreads: Effect.fn("ProviderSessionManager.otherWorkingThreads")( function* (input) { const entry = (yield* Ref.get(sessions)).get(sessionKey(input.providerSessionId)); @@ -2486,6 +2652,14 @@ export const layerWithOptions = ( }); if (Option.isSome(detachResult)) { yield* cancelIdleFiber(detachResult.value.idleUnloadFiber); + // A detached thread's run gets no terminal from this session, so it stops counting here. + for (const [providerThreadId, run] of runActivity) { + if ( + run.providerSessionId === input.providerSessionId && + run.threadId === input.threadId + ) + runActivity.delete(providerThreadId); + } } const detached = Option.map(detachResult, (result) => result.entry); // Plain detaches preserve credentials held by live shared providers across workspace handoffs. diff --git a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts index 7932a03809c94..a0258d1c8cc4d 100644 --- a/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderTurnControlService.test.ts @@ -288,6 +288,7 @@ it.effect( detach: () => Effect.void, otherWorkingThreads: () => Effect.succeed([]), releaseUnlessBusy: () => Effect.succeed(true), + quietRuns: Stream.empty, }), ); const layerControl = ProviderTurnControlService.layer.pipe( diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 898ac705084cf..1fec3d6e4bacf 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -3174,6 +3174,16 @@ export const layerWsRpc = ( })), ) : Stream.empty; + const quietRunUpdates = + input.quietRuns === true + ? providerSessionManager.quietRuns.pipe( + Stream.map((runs) => ({ + version: 1 as const, + type: "quietRunsUpdated" as const, + payload: { runs }, + })), + ) + : Stream.empty; const settingsUpdates = settingsChanges.pipe( Stream.map((settings) => ServerSettings.redactServerSettingsForClient(settings)), Stream.map((settings) => ({ @@ -3191,7 +3201,10 @@ export const layerWsRpc = ( settingsUpdates, Stream.merge( environmentThemeUpdates, - Stream.merge(usageLimitSourceUpdates, memoryPressureUpdates), + Stream.merge( + usageLimitSourceUpdates, + Stream.merge(memoryPressureUpdates, quietRunUpdates), + ), ), ), ), diff --git a/apps/web/src/components/chat/ThreadActivityBanners.tsx b/apps/web/src/components/chat/ThreadActivityBanners.tsx index 8bd26e21a04dc..7ed9673f13ff7 100644 --- a/apps/web/src/components/chat/ThreadActivityBanners.tsx +++ b/apps/web/src/components/chat/ThreadActivityBanners.tsx @@ -15,6 +15,7 @@ import * as DateTime from "effect/DateTime"; import { CircleDashedIcon, CircleStopIcon, + HourglassIcon, GitBranchIcon, MessageCircleIcon, ShieldQuestionIcon, @@ -148,6 +149,49 @@ export function runStopNowItem(input: { }; } +/** A running turn the provider has been silent on; tools can legitimately run this long. */ +export function quietRunNowItem(input: { + readonly threadId: ThreadId; + readonly quietSince: string; + readonly stopDisabled: boolean; + readonly stopping: boolean; + readonly onStop: () => void; + readonly canForceStop: boolean; + readonly forceStopping: boolean; + readonly onForceStop: () => void; +}): ThreadNowItem { + const quietSinceMs = Date.parse(input.quietSince); + return { + id: `quiet-run:${input.threadId}`, + icon: , + title: ( + <> + No activity for{" "} + formatDuration(Math.max(1_000, nowMs - quietSinceMs)), + true, + )} + className="tabular-nums" + /> + + ), + description: "A long tool may still be running, or the agent may be stuck.", + actions: ( + <> + + {input.stopping ? "Stopping..." : "Stop"} + + {input.canForceStop ? ( + + {input.forceStopping ? "Force stopping..." : "Force stop"} + + ) : null} + + ), + }; +} + /** Background commands and subagents, for the end of the timeline. */ export function backgroundWorkNowItem(input: { readonly threadId: ThreadId; diff --git a/apps/web/src/components/chat/useThreadLifecycle.tsx b/apps/web/src/components/chat/useThreadLifecycle.tsx index 53c33b5b1ada6..febddb5275c98 100644 --- a/apps/web/src/components/chat/useThreadLifecycle.tsx +++ b/apps/web/src/components/chat/useThreadLifecycle.tsx @@ -62,8 +62,10 @@ import { backgroundWorkNowItem, childInputNowItem, goalNowItem, + quietRunNowItem, runStopNowItem, } from "./ThreadActivityBanners"; +import { useQuietRunId, useQuietRunSince } from "../../hooks/useQuietRunSince"; import { useForceStopThread } from "./useForceStopThread"; import { useAtomValue } from "@effect/atom-react"; import type { RunStopPresentation } from "@t3tools/client-runtime/state/thread-execution"; @@ -541,6 +543,42 @@ export function useThreadLifecycle(input: ThreadLifecycleInput) { onForceStop: () => void forceStop.forceStopThread(activeThread.id, runStop.runId), }) : null; + const quietRunThread = isWorking && activeThread ? activeThreadShell : null; + const quietSince = useQuietRunSince(environmentId, quietRunThread); + const [stoppingQuietRunKey, setStoppingQuietRunKey] = useState(null); + // The run the server saw go quiet, not the latest one: a queued follow-up can be newer. + const quietRunId = useQuietRunId(environmentId, quietRunThread); + const handleStopQuietRun = async () => { + if (!activeThread) return; + const requestKey = `${environmentId}:${activeThread.id}`; + setStoppingQuietRunKey(requestKey); + const result = await interruptThreadTurn({ + environmentId, + input: + quietRunId === undefined + ? { threadId: activeThread.id } + : { threadId: activeThread.id, runId: quietRunId }, + }); + setStoppingQuietRunKey((current) => (current === requestKey ? null : current)); + if (result._tag === "Failure" && !isAtomCommandInterrupted(result)) { + const error = squashAtomCommandFailure(result); + setThreadError(activeThread.id, error instanceof Error ? error.message : "Failed to stop."); + } + }; + const quietRunNow = + activeThread && quietSince !== null && runStop === null + ? quietRunNowItem({ + threadId: activeThread.id, + quietSince, + stopDisabled: + !canOperateThread || stoppingQuietRunKey === `${environmentId}:${activeThread.id}`, + stopping: stoppingQuietRunKey === `${environmentId}:${activeThread.id}`, + onStop: () => void handleStopQuietRun(), + canForceStop: forceStop.canForceStop, + forceStopping: forceStop.stoppingThreadId === activeThread.id, + onForceStop: () => void forceStop.forceStopThread(activeThread.id, quietRunId), + }) + : null; const backgroundWorkNow = activeThread ? backgroundWorkNowItem({ threadId: activeThread.id, @@ -626,7 +664,7 @@ export function useThreadLifecycle(input: ThreadLifecycleInput) { activeBranchMismatchKey, // A Stop that went unanswered needs the user; one in progress is the thread's work. attentionNow: runStop?.state === "unresponsive" ? runStopNow : childInputNow, - workNow: runStop?.state === "stopping" ? runStopNow : backgroundWorkNow, + workNow: runStop?.state === "stopping" ? runStopNow : (quietRunNow ?? backgroundWorkNow), childInputNow, goalNow, handleSwitchCheckoutToThread, diff --git a/apps/web/src/components/sidebar/SidebarThreadRow.tsx b/apps/web/src/components/sidebar/SidebarThreadRow.tsx index a111126342f4e..f2e6489485745 100644 --- a/apps/web/src/components/sidebar/SidebarThreadRow.tsx +++ b/apps/web/src/components/sidebar/SidebarThreadRow.tsx @@ -64,6 +64,7 @@ import { useOpenPrLink } from "../../lib/openPullRequestLink"; import { useUiStateStore } from "../../uiStateStore"; import { useThreadSelectionStore } from "../../threadSelectionStore"; import { useNowMinute } from "../../hooks/useNowMinute"; +import { useQuietRunSince } from "../../hooks/useQuietRunSince"; import { useNowSecond } from "../../hooks/useNowSecond"; import { vcsEnvironment } from "../../state/vcs"; import { useEnvironmentQuery } from "../../state/query"; @@ -707,6 +708,7 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP const isUnread = hasUnseenCompletion({ ...thread, lastVisitedAt }); const status = resolveSidebarThreadStatus(thread); + const quietSince = useQuietRunSince(thread.environmentId, status === "working" ? thread : null); const lastVisitedDate = lastVisitedAt === undefined ? null : parseTimestampDate(lastVisitedAt); const wokeAtDate = props.wokeAt === null ? null : parseTimestampDate(props.wokeAt); const isWoke = @@ -722,11 +724,17 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP }); const topStatus = status === "working" - ? { - label: thread.goal?.status === "active" ? "Goal" : "Working", - icon: "working" as const, - className: "text-info-foreground", - } + ? quietSince !== null + ? { + label: "No activity", + icon: "waiting" as const, + className: "text-warning-foreground", + } + : { + label: thread.goal?.status === "active" ? "Goal" : "Working", + icon: "working" as const, + className: "text-info-foreground", + } : status === "waiting" ? { label: "Waiting", @@ -1489,7 +1497,9 @@ const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThreadRowP {topStatus.label} {status === "working" ? ( - + ) : status === "waiting" ? ( diff --git a/apps/web/src/connection/runtime.ts b/apps/web/src/connection/runtime.ts index 963e35de8d094..896345ab9b5db 100644 --- a/apps/web/src/connection/runtime.ts +++ b/apps/web/src/connection/runtime.ts @@ -38,6 +38,7 @@ const layerProvidedClientConnection = layerSnapshotLoader.pipe( environmentThemes: true, usageLimitSources: true, memoryPressure: true, + quietRuns: true, usageLimitsCommand: true, }), ), diff --git a/apps/web/src/hooks/useQuietRunSince.ts b/apps/web/src/hooks/useQuietRunSince.ts new file mode 100644 index 0000000000000..c8b47fbe62af6 --- /dev/null +++ b/apps/web/src/hooks/useQuietRunSince.ts @@ -0,0 +1,30 @@ +import { useAtomValue } from "@effect/atom-react"; +import { + quietRunId, + quietRunSince, + type QuietRunThread, +} from "@t3tools/client-runtime/state/thread-execution"; +import type { EnvironmentId, RunId } from "@t3tools/contracts"; + +import { serverEnvironment } from "../state/server"; + +/** Selects one thread's entry, so a quiet-runs update re-renders only the rows it changes. */ +export function useQuietRunSince( + environmentId: EnvironmentId, + thread: QuietRunThread | null, +): string | null { + return useAtomValue(serverEnvironment.configValueAtom(environmentId), (config) => + thread === null ? null : quietRunSince(config, thread), + ); +} + +/** The quiet run's id, for Stop and Force stop to target it and nothing newer. */ +export function useQuietRunId( + environmentId: EnvironmentId, + thread: QuietRunThread | null, +): RunId | undefined { + const runId = useAtomValue(serverEnvironment.configValueAtom(environmentId), (config) => + thread === null ? null : quietRunId(config, thread), + ); + return runId ?? undefined; +} diff --git a/apps/web/src/state/server.ts b/apps/web/src/state/server.ts index 12847c01dbd8c..59f67d5736ebd 100644 --- a/apps/web/src/state/server.ts +++ b/apps/web/src/state/server.ts @@ -36,6 +36,7 @@ export const serverEnvironment = createServerEnvironmentAtoms(connectionAtomRunt environmentThemes: true, usageLimitSources: true, memoryPressure: true, + quietRuns: true, usageLimitsCommand: true, }); export const environmentUpdateStatesAtom = environmentAtomList( diff --git a/packages/client-runtime/src/rpc/session.test.ts b/packages/client-runtime/src/rpc/session.test.ts index 8d0794fa234f7..96449c5480c39 100644 --- a/packages/client-runtime/src/rpc/session.test.ts +++ b/packages/client-runtime/src/rpc/session.test.ts @@ -504,112 +504,117 @@ describe("RpcSessionFactory", () => { ), ); - it.effect("replays theme updates, deletion and memory pressure as authoritative events", () => - Effect.scoped( - Effect.gen(function* () { - const { factory, sockets } = yield* makeFactory({ environmentThemes: true }); - const session = yield* factory.connect(PREPARED); - const readyFiber = yield* Effect.forkChild(session.ready); - const socket = yield* awaitSocket(sockets); - socket.open(); - yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, { - environmentThemes: true, - }); - yield* Fiber.join(readyFiber); + it.effect( + "replays theme updates, deletion, memory pressure and quiet runs as authoritative events", + () => + Effect.scoped( + Effect.gen(function* () { + const { factory, sockets } = yield* makeFactory({ environmentThemes: true }); + const session = yield* factory.connect(PREPARED); + const readyFiber = yield* Effect.forkChild(session.ready); + const socket = yield* awaitSocket(sockets); + socket.open(); + yield* completeInitialConfig(socket, ENCODED_THEME_SERVER_CONFIG, { + environmentThemes: true, + }); + yield* Fiber.join(readyFiber); - const firstThemes = [ - { - id: "nightfall", - name: "Nightfall", - appearance: "dark" as const, - canvas: "#1a1b26", - accent: "#7aa2f7", - }, - ]; - const replacementThemes = [ - { - id: "midnight", - name: "Midnight", - appearance: "dark" as const, - canvas: "#000000", - accent: "#ffffff", - }, - ]; - const subscriberStarted = yield* Deferred.make(); - const subscriber = yield* session.subscribeServerConfig({ environmentThemes: true }).pipe( - Stream.mapEffect((event) => - Deferred.succeed(subscriberStarted, undefined).pipe(Effect.as(event)), - ), - Stream.take(5), - Stream.runCollect, - Effect.forkChild, - ); - yield* Deferred.await(subscriberStarted); - yield* publishConfigEvents(socket, [ - { - version: 1, - type: "memoryPressureUpdated", - payload: { - level: "severe", - heaviestAgents: [], + const firstThemes = [ + { + id: "nightfall", + name: "Nightfall", + appearance: "dark" as const, + canvas: "#1a1b26", + accent: "#7aa2f7", }, - }, - { - version: 1, - type: "environmentThemesUpdated", - payload: { themes: firstThemes }, - }, - { - version: 1, - type: "environmentThemesUpdated", - payload: { themes: replacementThemes }, - }, - { - version: 1, - type: "environmentThemesUpdated", - payload: { themes: [] }, - }, - ]); + ]; + const replacementThemes = [ + { + id: "midnight", + name: "Midnight", + appearance: "dark" as const, + canvas: "#000000", + accent: "#ffffff", + }, + ]; + const subscriberStarted = yield* Deferred.make(); + const subscriber = yield* session.subscribeServerConfig({ environmentThemes: true }).pipe( + Stream.mapEffect((event) => + Deferred.succeed(subscriberStarted, undefined).pipe(Effect.as(event)), + ), + Stream.take(6), + Stream.runCollect, + Effect.forkChild, + ); + yield* Deferred.await(subscriberStarted); + yield* publishConfigEvents(socket, [ + { + version: 1, + type: "memoryPressureUpdated", + payload: { + level: "severe", + heaviestAgents: [], + }, + }, + { version: 1, type: "quietRunsUpdated", payload: { runs: [] } }, + { + version: 1, + type: "environmentThemesUpdated", + payload: { themes: firstThemes }, + }, + { + version: 1, + type: "environmentThemesUpdated", + payload: { themes: replacementThemes }, + }, + { + version: 1, + type: "environmentThemesUpdated", + payload: { themes: [] }, + }, + ]); - const liveEvents = Array.from(yield* Fiber.join(subscriber)); - expect(liveEvents.map((event) => event.type)).toEqual([ - "snapshot", - "memoryPressureUpdated", - "environmentThemesUpdated", - "environmentThemesUpdated", - "environmentThemesUpdated", - ]); - expect(liveEvents[3]).toMatchObject({ payload: { themes: replacementThemes } }); + const liveEvents = Array.from(yield* Fiber.join(subscriber)); + expect(liveEvents.map((event) => event.type)).toEqual([ + "snapshot", + "memoryPressureUpdated", + "quietRunsUpdated", + "environmentThemesUpdated", + "environmentThemesUpdated", + "environmentThemesUpdated", + ]); + expect(liveEvents[4]).toMatchObject({ payload: { themes: replacementThemes } }); - // A late subscriber also needs the live pressure, which snapshots never carry. - const replay = Array.from( - yield* session - .subscribeServerConfig({ environmentThemes: true }) - .pipe(Stream.take(3), Stream.runCollect), - ); - expect(replay.map((event) => event.type)).toEqual([ - "snapshot", - "environmentThemesUpdated", - "memoryPressureUpdated", - ]); - expect(replay[1]).toMatchObject({ payload: { themes: [] } }); + // A late subscriber also needs live pressure and quiet runs, which snapshots never carry. + const replay = Array.from( + yield* session + .subscribeServerConfig({ environmentThemes: true }) + .pipe(Stream.take(4), Stream.runCollect), + ); + expect(replay.map((event) => event.type)).toEqual([ + "snapshot", + "environmentThemesUpdated", + "memoryPressureUpdated", + "quietRunsUpdated", + ]); + expect(replay[1]).toMatchObject({ payload: { themes: [] } }); - let projection = applyServerConfigProjection(Option.none(), { - version: 1, - type: "snapshot", - config: THEME_SERVER_CONFIG, - }); - projection = applyServerConfigProjection(projection, { - version: 1, - type: "environmentThemesUpdated", - payload: { themes: firstThemes }, - }); - for (const event of replay) { - projection = applyServerConfigProjection(projection, event); - } - expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined(); - }), - ), + let projection = applyServerConfigProjection(Option.none(), { + version: 1, + type: "snapshot", + config: THEME_SERVER_CONFIG, + }); + projection = applyServerConfigProjection(projection, { + version: 1, + type: "environmentThemesUpdated", + payload: { themes: firstThemes }, + }); + for (const event of replay) { + projection = applyServerConfigProjection(projection, event); + } + expect(Option.getOrThrow(projection).config.environmentThemes).toBeUndefined(); + }), + ), ); it.effect("recovers a slow subscriber after it misses theme and usage-source deletion", () => diff --git a/packages/client-runtime/src/rpc/session.ts b/packages/client-runtime/src/rpc/session.ts index 6eff643ff5cf2..02c3822a1b7d1 100644 --- a/packages/client-runtime/src/rpc/session.ts +++ b/packages/client-runtime/src/rpc/session.ts @@ -62,6 +62,7 @@ export interface RpcSessionOptions { readonly environmentThemes?: boolean; readonly usageLimitSources?: boolean; readonly memoryPressure?: boolean; + readonly quietRuns?: boolean; readonly usageLimitsCommand?: boolean; } @@ -100,6 +101,10 @@ type MemoryPressureUpdatedEvent = Extract< ServerConfigStreamEvent, { readonly type: "memoryPressureUpdated" } >; +type QuietRunsUpdatedEvent = Extract< + ServerConfigStreamEvent, + { readonly type: "quietRunsUpdated" } +>; interface ServerConfigReplayState { readonly projection: ServerConfigProjection; @@ -107,6 +112,7 @@ interface ServerConfigReplayState { readonly themesEvent: EnvironmentThemesUpdatedEvent | undefined; readonly sourcesEvent: UsageLimitSourcesUpdatedEvent | undefined; readonly pressureEvent: MemoryPressureUpdatedEvent | undefined; + readonly quietRunsEvent: QuietRunsUpdatedEvent | undefined; } interface BufferedServerConfigEvent { @@ -128,6 +134,7 @@ function serverConfigReplayEvents( ...(state.themesEvent === undefined ? [] : [state.themesEvent]), ...(state.sourcesEvent === undefined ? [] : [state.sourcesEvent]), ...(state.pressureEvent === undefined ? [] : [state.pressureEvent]), + ...(state.quietRunsEvent === undefined ? [] : [state.quietRunsEvent]), ]; } @@ -178,6 +185,7 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( ...(options.environmentThemes === true ? { environmentThemes: true } : {}), ...(options.usageLimitSources === true ? { usageLimitSources: true } : {}), ...(options.memoryPressure === true ? { memoryPressure: true } : {}), + ...(options.quietRuns === true ? { quietRuns: true } : {}), ...(options.usageLimitsCommand === true ? { usageLimitsCommand: true } : {}), }; // Config state merges the snapshots it fetches back in, so no subscriber needs them broadcast. @@ -305,6 +313,10 @@ export const make = Effect.fn("RpcSessionFactory.make")(function* ( event.type === "memoryPressureUpdated" ? event : Option.getOrUndefined(current)?.pressureEvent, + quietRunsEvent: + event.type === "quietRunsUpdated" + ? event + : Option.getOrUndefined(current)?.quietRunsEvent, } satisfies ServerConfigReplayState; return [ Option.some({ event, replay: next, revision: next.revision }), diff --git a/packages/client-runtime/src/state/server.test.ts b/packages/client-runtime/src/state/server.test.ts index f712a9c31c24a..bb17a6a4b1b52 100644 --- a/packages/client-runtime/src/state/server.test.ts +++ b/packages/client-runtime/src/state/server.test.ts @@ -2,7 +2,9 @@ import { EnvironmentId, ServerSelfUpdateError, ProviderInstanceId, + RunId, ServerObservability, + ThreadId, EnvironmentThemeId, UsageLimitSourceId, type ServerConfig, @@ -13,6 +15,7 @@ import { import { describe, expect, it } from "@effect/vitest"; import * as Cause from "effect/Cause"; import * as Deferred from "effect/Deferred"; +import * as DateTime from "effect/DateTime"; import * as Duration from "effect/Duration"; import * as Effect from "effect/Effect"; import * as Exit from "effect/Exit"; @@ -1294,6 +1297,13 @@ describe("server state projection", () => { level: "severe", heaviestAgents: [], }, + quietRuns: [ + { + threadId: ThreadId.make("thread-quiet"), + runId: RunId.make("run-quiet"), + quietSince: DateTime.makeUnsafe(0), + }, + ], }, }); const projected = yield* SubscriptionRef.changes(state).pipe( @@ -1316,6 +1326,7 @@ describe("server state projection", () => { expect(saved).not.toHaveProperty("lastEventLoopStall"); // Never read back, and changes every sample while memory is short. expect(saved).not.toHaveProperty("memoryPressure"); + expect(saved).not.toHaveProperty("quietRuns"); }), ); }); diff --git a/packages/client-runtime/src/state/server.ts b/packages/client-runtime/src/state/server.ts index bde841628d58f..6c2ce4fa68388 100644 --- a/packages/client-runtime/src/state/server.ts +++ b/packages/client-runtime/src/state/server.ts @@ -365,6 +365,7 @@ export interface ServerConfigSubscriptionOptions { readonly environmentThemes?: boolean; readonly usageLimitSources?: boolean; readonly memoryPressure?: boolean; + readonly quietRuns?: boolean; readonly usageLimitsCommand?: boolean; } @@ -433,12 +434,13 @@ export const makeEnvironmentServerConfigState = Effect.fn("EnvironmentServerConf const persist = Effect.fn("EnvironmentServerConfigState.persist")(function* ( config: ServerConfig, ) { - // A cached stall would read as a fresh freeze on the next launch. Pressure is live-only. + // A cached stall would read as a fresh freeze on the next launch. Pressure and quiet runs are live-only. const { environmentThemes: _themes, usageLimitSources: _sources, lastEventLoopStall: _stall, memoryPressure: _pressure, + quietRuns: _quietRuns, ...cacheable } = config; if (lastPersisted !== undefined && sameCacheValue(lastPersisted, cacheable)) { @@ -488,6 +490,7 @@ export const makeEnvironmentServerConfigState = Effect.fn("EnvironmentServerConf ...(subscription.environmentThemes === true ? { environmentThemes: true } : {}), ...(subscription.usageLimitSources === true ? { usageLimitSources: true } : {}), ...(subscription.memoryPressure === true ? { memoryPressure: true } : {}), + ...(subscription.quietRuns === true ? { quietRuns: true } : {}), ...(subscription.usageLimitsCommand === true ? { usageLimitsCommand: true } : {}), }).pipe( Stream.runForEach((event) => @@ -758,6 +761,7 @@ export function createServerEnvironmentAtoms( readonly environmentThemes?: boolean; readonly usageLimitSources?: boolean; readonly memoryPressure?: boolean; + readonly quietRuns?: boolean; readonly usageLimitsCommand?: boolean; }, ) { @@ -774,6 +778,7 @@ export function createServerEnvironmentAtoms( ...(options.environmentThemes === true ? { environmentThemes: true } : {}), ...(options.usageLimitSources === true ? { usageLimitSources: true } : {}), ...(options.memoryPressure === true ? { memoryPressure: true } : {}), + ...(options.quietRuns === true ? { quietRuns: true } : {}), ...(options.usageLimitsCommand === true ? { usageLimitsCommand: true } : {}), }), ) diff --git a/packages/client-runtime/src/state/serverConfigProjection.test.ts b/packages/client-runtime/src/state/serverConfigProjection.test.ts index fb928b83cda6a..3d6da5186a504 100644 --- a/packages/client-runtime/src/state/serverConfigProjection.test.ts +++ b/packages/client-runtime/src/state/serverConfigProjection.test.ts @@ -56,18 +56,22 @@ describe("applyServerConfigProjection settingsUpdated", () => { }); describe("applyServerConfigProjection memoryPressureUpdated", () => { - it("keeps pressure across a reconnect snapshot from a server that sends it, and drops it once pressure ends", () => { + it("keeps pressure and quiet runs across a reconnect snapshot from a server that sends them, and drops them once they end", () => { const pressure = (level: "none" | "severe") => ({ type: "memoryPressureUpdated", version: 1, payload: { level, heaviestAgents: [] }, }) as ServerConfigStreamEvent; - const snapshotFrom = (memoryPressure: boolean) => + const snapshotFrom = (sendsEvents: boolean) => ({ type: "snapshot", version: 1, - config: { environment: { capabilities: { memoryPressure } } }, + config: { + environment: { + capabilities: { memoryPressure: sendsEvents, quietRuns: sendsEvents }, + }, + }, }) as ServerConfigStreamEvent; const snapshot = snapshotFrom(true); const calm = applyServerConfigProjection( @@ -75,15 +79,30 @@ describe("applyServerConfigProjection memoryPressureUpdated", () => { pressure("none"), ); expect(Option.isNone(calm)).toBe(true); + const quietRuns = (threadIds: ReadonlyArray) => + ({ + type: "quietRunsUpdated", + version: 1, + payload: { runs: threadIds.map((threadId) => ({ threadId, quietSince: 0 })) }, + }) as unknown as ServerConfigStreamEvent; const severe = applyServerConfigProjection( - applyServerConfigProjection(Option.none(), snapshot), - pressure("severe"), + applyServerConfigProjection( + applyServerConfigProjection(Option.none(), snapshot), + pressure("severe"), + ), + quietRuns(["thread-a"]), + ); + const reconnected = Option.getOrThrow(applyServerConfigProjection(severe, snapshot)); + expect(reconnected.config.memoryPressure?.level).toBe("severe"); + expect(reconnected.config.quietRuns?.map((run) => run.threadId)).toEqual(["thread-a"]); + const olderServer = Option.getOrThrow(applyServerConfigProjection(severe, snapshotFrom(false))); + expect(olderServer.config.memoryPressure).toBeUndefined(); + expect(olderServer.config.quietRuns).toBeUndefined(); + const ended = applyServerConfigProjection( + applyServerConfigProjection(Option.some(reconnected), pressure("none")), + quietRuns([]), ); - const reconnected = applyServerConfigProjection(severe, snapshot); - expect(Option.getOrThrow(reconnected).config.memoryPressure?.level).toBe("severe"); - const olderServer = applyServerConfigProjection(severe, snapshotFrom(false)); - expect(Option.getOrThrow(olderServer).config.memoryPressure).toBeUndefined(); - const ended = applyServerConfigProjection(reconnected, pressure("none")); expect(Option.getOrThrow(ended).config.memoryPressure).toBeUndefined(); + expect(Option.getOrThrow(ended).config.quietRuns).toBeUndefined(); }); }); diff --git a/packages/client-runtime/src/state/serverConfigProjection.ts b/packages/client-runtime/src/state/serverConfigProjection.ts index 58032fc35e939..de9eb82430e93 100644 --- a/packages/client-runtime/src/state/serverConfigProjection.ts +++ b/packages/client-runtime/src/state/serverConfigProjection.ts @@ -15,7 +15,8 @@ export function withoutEnvironmentThemes(config: ServerConfig): ServerConfig { if ( config.environmentThemes === undefined && config.usageLimitSources === undefined && - config.memoryPressure === undefined + config.memoryPressure === undefined && + config.quietRuns === undefined ) { return config; } @@ -23,6 +24,7 @@ export function withoutEnvironmentThemes(config: ServerConfig): ServerConfig { environmentThemes: _themes, usageLimitSources: _sources, memoryPressure: _pressure, + quietRuns: _quietRuns, ...rest } = config; return rest; @@ -81,13 +83,19 @@ export function applyServerConfigProjection( const carriedSources = Option.isSome(current) ? current.value.config.usageLimitSources : undefined; - // A server without the event would never clear carried pressure. + // A server without the event would never clear carried pressure or quiet runs. + const capabilities = event.config.environment.capabilities; const carriedPressure = - Option.isSome(current) && event.config.environment.capabilities.memoryPressure === true + Option.isSome(current) && capabilities.memoryPressure === true ? current.value.config.memoryPressure : undefined; + const carriedQuietRuns = + Option.isSome(current) && capabilities.quietRuns === true + ? current.value.config.quietRuns + : undefined; const config: ServerConfig = { ...event.config, + ...(carriedQuietRuns === undefined ? {} : { quietRuns: carriedQuietRuns }), ...(carriedThemes === undefined ? {} : { environmentThemes: carriedThemes }), ...(carriedSources === undefined ? {} : { usageLimitSources: carriedSources }), ...(carriedPressure === undefined ? {} : { memoryPressure: carriedPressure }), @@ -154,6 +162,16 @@ export function applyServerConfigProjection( source: "live", }); }); + case "quietRunsUpdated": + return Option.map(current, (projection) => { + const { quietRuns: _previous, ...config } = projection.config; + return { + config: + event.payload.runs.length === 0 ? config : { ...config, quietRuns: event.payload.runs }, + latestEvent: event, + source: "live", + }; + }); case "usageLimitSourcesUpdated": return Option.map(current, (projection) => ({ config: { diff --git a/packages/client-runtime/src/state/threadExecution.test.ts b/packages/client-runtime/src/state/threadExecution.test.ts index 81afabbe8aa49..8129b486502fa 100644 --- a/packages/client-runtime/src/state/threadExecution.test.ts +++ b/packages/client-runtime/src/state/threadExecution.test.ts @@ -24,6 +24,9 @@ import { deriveProviderSubagentStatus, deriveRunlessWorkStartedAt, deriveThreadActivityRun, + quietRunId, + quietRunSince, + type QuietRunThread, deriveThreadRuntime, presentProviderGoal, presentRunStop, @@ -488,3 +491,37 @@ describe("presentProviderGoal", () => { ).toBe("1m / 1m tokens"); }); }); + +describe("quiet run selectors", () => { + const threadId = ThreadId.make("thread-quiet"); + const quietRun = RunId.make("run-quiet"); + const config = { + quietRuns: [{ threadId, runId: quietRun, quietSince: DateTime.makeUnsafe(0) }], + }; + + it("flag the quiet run only while it is still the active run and not waiting", () => { + const thread = (activeRunId: RunId | null, waiting: Partial = {}) => ({ + id: threadId, + runtime: { activeRunId }, + hasPendingApprovals: false, + hasPendingUserInput: false, + ...waiting, + }); + expect(quietRunId(config, thread(quietRun))).toBe(quietRun); + expect(quietRunSince(config, thread(quietRun))).toBe("1970-01-01T00:00:00.000Z"); + // A follow-up started before the next quiet-runs update. + const followUp = RunId.make("run-follow-up"); + expect(quietRunId(config, thread(followUp))).toBeNull(); + expect(quietRunSince(config, thread(followUp))).toBeNull(); + expect(quietRunSince(config, thread(null))).toBeNull(); + for (const waiting of [ + { hasPendingApprovals: true }, + { hasPendingUserInput: true }, + { pendingSubagentRequest: "approval" }, + { blockedOnThreadWait: true }, + ]) { + expect(quietRunId(config, thread(quietRun, waiting))).toBeNull(); + expect(quietRunSince(config, thread(quietRun, waiting))).toBeNull(); + } + }); +}); diff --git a/packages/client-runtime/src/state/threadExecution.ts b/packages/client-runtime/src/state/threadExecution.ts index 951e5d06404b3..0584a4e1a8647 100644 --- a/packages/client-runtime/src/state/threadExecution.ts +++ b/packages/client-runtime/src/state/threadExecution.ts @@ -12,6 +12,8 @@ import { type OrchestrationV2NotificationSource, type OrchestrationV2PendingBackgroundTask, type OrchestrationV2ProviderGoal, + type RunId, + type ServerConfig, type ServerProviderModel, type OrchestrationV2ExecutionNode, type OrchestrationV2ThreadProjection, @@ -343,6 +345,52 @@ export function presentRunStop( }; } +// The set refreshes every 30 s, so a follow-up can start before it drops a finished run. +export interface QuietRunThread { + readonly id: ThreadId; + readonly runtime: Pick | null; + readonly hasPendingApprovals: boolean; + readonly hasPendingUserInput: boolean; + readonly pendingSubagentRequest?: unknown; + readonly blockedOnThreadWait?: boolean | undefined; +} + +// A run waiting on the user or another thread is silent by design, not stuck. +const activeQuietRun = ( + config: Pick | null | undefined, + thread: QuietRunThread, +) => { + const activeRunId = thread.runtime?.activeRunId; + if ( + activeRunId == null || + thread.hasPendingApprovals || + thread.hasPendingUserInput || + thread.pendingSubagentRequest != null || + thread.blockedOnThreadWait === true + ) + return undefined; + return config?.quietRuns?.find( + (candidate) => candidate.threadId === thread.id && candidate.runId === activeRunId, + ); +}; + +/** The quiet run itself, so Stop and Force stop cannot reach a newer one. */ +export function quietRunId( + config: Pick | null | undefined, + thread: QuietRunThread, +): RunId | null { + return activeQuietRun(config, thread)?.runId ?? null; +} + +/** When the active run last heard from its provider, once that is long enough ago to flag. */ +export function quietRunSince( + config: Pick | null | undefined, + thread: QuietRunThread, +): string | null { + const run = activeQuietRun(config, thread); + return run === undefined ? null : DateTime.formatIso(run.quietSince); +} + export interface PendingBackgroundWorkItem { readonly taskId: string; readonly kind: BackgroundWorkKind; diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index f40968f6a36b4..90ecabac55244 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -103,6 +103,8 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ threadForceStop: Schema.optionalKey(Schema.Boolean), /** Sends `memoryPressureUpdated` to config subscribers that ask for it. */ memoryPressure: Schema.optionalKey(Schema.Boolean), + /** Sends `quietRunsUpdated` to config subscribers that ask for it. */ + quietRuns: Schema.optionalKey(Schema.Boolean), /** Every protocol-2 server sends this; older mobile builds still read it. */ environmentThemes: Schema.optionalKey(Schema.Boolean), usageLimitSources: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/rpc.ts b/packages/contracts/src/rpc.ts index a4b444e1927be..04f1d844a49af 100644 --- a/packages/contracts/src/rpc.ts +++ b/packages/contracts/src/rpc.ts @@ -1846,6 +1846,7 @@ export const WsSubscribeServerConfigRpc = Rpc.make(WS_METHODS.subscribeServerCon environmentThemes: Schema.optional(Schema.Boolean), usageLimitSources: Schema.optional(Schema.Boolean), memoryPressure: Schema.optional(Schema.Boolean), + quietRuns: Schema.optional(Schema.Boolean), /** Advertise client handling before the server offers /usage-limits, or older clients send it as a prompt. */ usageLimitsCommand: Schema.optional(Schema.Boolean), /** Omit per-cwd provider snapshots; the client fetches its cwd through `server.refreshProviders`. */ diff --git a/packages/contracts/src/server.ts b/packages/contracts/src/server.ts index 3f9bb6ba306c5..e4d47a6e90e26 100644 --- a/packages/contracts/src/server.ts +++ b/packages/contracts/src/server.ts @@ -13,6 +13,7 @@ import { NonNegativeInt, PositiveInt, ProjectId, + RunId, ThreadId, TrimmedNonEmptyString, } from "./baseSchemas.ts"; @@ -607,6 +608,15 @@ export const ServerEventLoopStall = Schema.Struct({ }); export type ServerEventLoopStall = typeof ServerEventLoopStall.Type; +/** A running turn whose provider has sent nothing since `quietSince`. */ +export const ServerQuietRun = Schema.Struct({ + threadId: ThreadId, + /** The quiet run itself, so Stop and Force stop cannot reach a newer one. */ + runId: RunId, + quietSince: Schema.DateTimeUtc, +}); +export type ServerQuietRun = typeof ServerQuietRun.Type; + export const ServerConfig = Schema.Struct({ environment: ExecutionEnvironmentDescriptor, auth: ServerAuthDescriptor, @@ -661,6 +671,8 @@ export const ServerConfig = Schema.Struct({ usageLimitSources: Schema.optional(UsageLimitSourceSnapshots), /** Stream-only and opt-in, like `usageLimitSources`; absent while there is none. */ memoryPressure: Schema.optional(HostMemoryPressure), + /** Stream-only and opt-in; turns quiet for ten minutes or more. */ + quietRuns: Schema.optional(Schema.Array(ServerQuietRun)), }); export type ServerConfig = typeof ServerConfig.Type; @@ -790,6 +802,14 @@ export const ServerConfigStreamMemoryPressureUpdatedEvent = Schema.Struct({ export type ServerConfigStreamMemoryPressureUpdatedEvent = typeof ServerConfigStreamMemoryPressureUpdatedEvent.Type; +export const ServerConfigStreamQuietRunsUpdatedEvent = Schema.Struct({ + version: Schema.Literal(1), + type: Schema.Literal("quietRunsUpdated"), + payload: Schema.Struct({ runs: Schema.Array(ServerQuietRun) }), +}); +export type ServerConfigStreamQuietRunsUpdatedEvent = + typeof ServerConfigStreamQuietRunsUpdatedEvent.Type; + export const ServerConfigStreamEvent = Schema.Union([ ServerConfigStreamSnapshotEvent, ServerConfigStreamKeybindingsUpdatedEvent, @@ -798,6 +818,7 @@ export const ServerConfigStreamEvent = Schema.Union([ ServerConfigStreamEnvironmentThemesUpdatedEvent, ServerConfigStreamUsageLimitSourcesUpdatedEvent, ServerConfigStreamMemoryPressureUpdatedEvent, + ServerConfigStreamQuietRunsUpdatedEvent, ]); export type ServerConfigStreamEvent = typeof ServerConfigStreamEvent.Type;