authorgravatar for me@t3.ggTheo Browne <me@t3.gg> 2026-10-02 21:55:57-07:00
committergravatar for noreply@github.comGitHub <noreply@github.com> 2026-10-02 21:55:57-07:00
log736130c2fd39b85cd515fa2a41cd10c8a8a52630
tree8711cb01b83d3a765cc77d830c7d8e317ab7fdd0
parent9333509c918083cbfb7e66759ddbcbfa71a3a452
signature Signed by PGP key B5690EEEBB952194

fix(server): working timers no longer reset on every background wake (#15029)

Co-authored-by: Claude Opus 5.5 (1M context) <noreply@anthropic.com>

9 files changed, 245 insertions(+), 11 deletions(-)

apps/server/src/orchestration-v2/Orchestrator.ts+39
......@@ -39,6 +39,7 @@ import {
3939 type OrchestrationV2Subagent,
4040 type OrchestrationV2ThreadProjection,
4141 type OrchestrationV2TurnItem,
42 orchestrationV2RunWorkStartedAt,
4243 ProviderInstanceId,
4344 type ProviderSessionId,
4445 RunId,
......@@ -304,6 +305,37 @@ function nextRunOrdinal(projection: Pick<OrchestrationV2ThreadProjection, "runs"
304305 return projection.runs.length + 1;
305306}
306307
308/**
309 * A wake (background notification, delegated task result, restart
310 * continuation) carries on the work of the run that started last, so it keeps
311 * that work's start. Stamp it when the wake run starts, not when it queues:
312 * a queued prompt ahead of it has no start yet, and delegated results jump
313 * the queue. Other runs start new work.
314 */
315function wakeWorkStartedAt(
316 runs: ReadonlyArray<OrchestrationV2Run>,
317 trigger: {
318 readonly notification?: unknown;
319 readonly delegatedCompletion?: unknown;
320 readonly restartContinuationOfRunId?: RunId | undefined;
321 },
322): Pick<OrchestrationV2Run, "workStartedAt"> {
323 if (
324 trigger.notification === undefined &&
325 trigger.delegatedCompletion === undefined &&
326 trigger.restartContinuationOfRunId === undefined
327 ) {
328 return {};
329 }
330 const previous = runs
331 .flatMap((run) => (run.startedAt === null ? [] : [{ run, startedAt: run.startedAt }]))
332 .toSorted(
333 (left, right) =>
334 DateTime.Order(right.startedAt, left.startedAt) || right.run.ordinal - left.run.ordinal,
335 )[0]?.run;
336 return previous === undefined ? {} : { workStartedAt: orchestrationV2RunWorkStartedAt(previous) };
337}
338
307339function isNativeMaintenanceCommand(message: {
308340 readonly text: string;
309341 readonly attachments: ReadonlyArray<ChatAttachment>;
......@@ -1459,6 +1491,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
14591491 queuePosition: null,
14601492 startedAt: null,
14611493 contextHandoffId: activeHandoff?.id ?? null,
1494 ...wakeWorkStartedAt(projection.runs, {
1495 notification: queuedMessage.notification,
1496 delegatedCompletion: queuedMessage.delegatedCompletion,
1497 restartContinuationOfRunId: queuedRun.restartContinuationOfRunId,
1498 }),
14621499 };
14631500 const userTurnItem: OrchestrationV2TurnItem = {
14641501 ...(legacyQueuedTurnItem ?? {
......@@ -4949,6 +4986,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
49494986 ...(command.restartContinuationOfRunId === undefined
49504987 ? {}
49514988 : { restartContinuationOfRunId: command.restartContinuationOfRunId }),
4989 ...wakeWorkStartedAt(projection.runs, command),
49524990 };
49534991 const attempt: OrchestrationV2RunAttempt = {
49544992 id: attemptId,
......@@ -5642,6 +5680,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio
56425680 ...(command.restartContinuationOfRunId === undefined
56435681 ? {}
56445682 : { restartContinuationOfRunId: command.restartContinuationOfRunId }),
5683 ...wakeWorkStartedAt(projection.runs, command),
56455684 };
56465685 const attempt: OrchestrationV2RunAttempt = {
56475686 id: attemptId,
apps/server/src/orchestration-v2/ProjectionStore.test.ts+31
......@@ -1982,6 +1982,37 @@ it.layer(TestLayer)("ProjectionStoreV2", (it) => {
19821982 assert.isNull(onlyHeldShell.latestRunId);
19831983 assert.equal(onlyHeldShell.status, "idle");
19841984 }
1985
1986 // A wake run counts from the start of the work it continues.
1987 yield* projectionStore.apply({
1988 id: EventId.make("event:projection-shell-interruptible:wake"),
1989 type: "run.updated",
1990 threadId,
1991 runId,
1992 nodeId: rootNodeId,
1993 driver,
1994 occurredAt: later,
1995 payload: {
1996 ...run,
1997 status: "running",
1998 requestedAt: later,
1999 startedAt: later,
2000 workStartedAt: now,
2001 },
2002 });
2003 const wakeProjection = yield* projectionStore.getThreadProjection(threadId);
2004 const wakeSqlShell = (yield* projectionStore.getShellSnapshot()).threads.find(
2005 (row) => row.id === threadId,
2006 )!;
2007 for (const wakeShell of [
2008 wakeSqlShell,
2009 ProjectionStore.threadShellFromProjection(wakeProjection),
2010 ]) {
2011 assert.equal(
2012 wakeShell.activityRunStartedAt && DateTime.toEpochMillis(wakeShell.activityRunStartedAt),
2013 DateTime.toEpochMillis(now),
2014 );
2015 }
19852016 }),
19862017 );
19872018
apps/server/src/orchestration-v2/ProjectionStore.ts+9-2
......@@ -48,6 +48,7 @@ import {
4848 OrchestrationV2RuntimeRequestJson as OrchestrationV2RuntimeRequestJsonSchema,
4949 OrchestrationV2SubagentJson as OrchestrationV2SubagentJsonSchema,
5050 OrchestrationV2TurnItemJson as OrchestrationV2TurnItemJsonSchema,
51 orchestrationV2RunWorkStartedAt,
5152 RunId,
5253 CheckpointScopeId,
5354 ThreadId,
......@@ -1368,7 +1369,8 @@ export function threadShellFromProjection(
13681369 latestRunCompletedAt: latestRun?.completedAt ?? null,
13691370 activeRunId: activeRun?.id ?? null,
13701371 activityRunStatus: activityRun?.status ?? null,
1371 activityRunStartedAt: activityRun?.startedAt ?? activityRun?.requestedAt ?? null,
1372 activityRunStartedAt:
1373 activityRun === null ? null : orchestrationV2RunWorkStartedAt(activityRun),
13721374 status: latestRun?.status ?? "idle",
13731375 ...threadErrorSummary(
13741376 latestRootProviderFailure(latestRun, projection.turnItems),
......@@ -4782,7 +4784,12 @@ export const layer: Layer.Layer<ProjectionStoreV2, never, SqlClient.SqlClient> =
47824784 LIMIT 1
47834785 ) AS activity_run_status,
47844786 (
4785 SELECT COALESCE(json_extract(r.payload_json, '$.startedAt'), r.requested_at)
4787 -- Mirrors orchestrationV2RunWorkStartedAt.
4788 SELECT COALESCE(
4789 json_extract(r.payload_json, '$.workStartedAt'),
4790 json_extract(r.payload_json, '$.startedAt'),
4791 r.requested_at
4792 )
47864793 FROM orchestration_v2_projection_runs r
47874794 WHERE r.thread_id = t.thread_id
47884795 AND r.status IN ('preparing', 'starting', 'running', 'waiting')
apps/server/src/orchestration-v2/runtimeLayer.test.ts+118
......@@ -2800,6 +2800,124 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => {
28002800 }),
28012801 );
28022802
2803 it.effect("starts a wake's work clock from the run that ran before it", () =>
2804 Effect.gen(function* () {
2805 const orchestrator = yield* Orchestrator.OrchestratorV2;
2806 const eventSink = yield* EventSink.EventSinkV2;
2807 const threadId = ThreadId.make("runtime-layer-wake-work-start-thread");
2808 const messageId = (key: string) => MessageId.make(`runtime-layer-wake-work-start-${key}`);
2809
2810 yield* orchestrator.dispatch({
2811 type: "thread.create",
2812 createdBy: "user",
2813 creationSource: "web",
2814 commandId: CommandId.make("runtime-layer-wake-work-start-create"),
2815 threadId,
2816 projectId: ProjectId.make("runtime-layer-wake-work-start-project"),
2817 title: "Wake work start",
2818 modelSelection,
2819 runtimeMode: "full-access",
2820 interactionMode: "default",
2821 branch: null,
2822 worktreePath: process.cwd(),
2823 });
2824 const dispatch = (key: string, wake: boolean) =>
2825 orchestrator.dispatch({
2826 type: "message.dispatch",
2827 createdBy: wake ? "agent" : "user",
2828 creationSource: wake ? "provider" : "web",
2829 ...(wake
2830 ? {
2831 notification: {
2832 source: { kind: "background_task" as const },
2833 outcome: "updated" as const,
2834 summary: "Background activity updated",
2835 },
2836 }
2837 : {}),
2838 commandId: CommandId.make(`runtime-layer-wake-work-start-${key}`),
2839 threadId,
2840 messageId: messageId(key),
2841 text: key,
2842 attachments: [],
2843 modelSelection,
2844 dispatchMode:
2845 key === "prompt" ? { type: "start_immediately" } : { type: "queue_after_active" },
2846 });
2847 const runFor = (key: string) =>
2848 Effect.map(orchestrator.getThreadProjection(threadId), ({ runs }) => {
2849 const run = runs.find((candidate) => candidate.userMessageId === messageId(key));
2850 assert.isDefined(run);
2851 return run;
2852 });
2853 yield* dispatch("prompt", false);
2854 yield* dispatch("queued", false);
2855 yield* dispatch("early-wake", true);
2856 // The early wake now runs ahead of the older queued prompt, as a
2857 // delegated result does when it jumps the queue.
2858 yield* orchestrator.dispatch({
2859 type: "queued-run.reorder",
2860 commandId: CommandId.make("runtime-layer-wake-work-start-reorder"),
2861 threadId,
2862 runId: (yield* runFor("queued")).id,
2863 beforeRunId: null,
2864 });
2865 yield* dispatch("late-wake", true);
2866 // A queued wake has no clock yet: what runs before it is still unknown.
2867 assert.isUndefined((yield* runFor("early-wake")).workStartedAt);
2868
2869 const startedRunIds = yield* Queue.unbounded<RunId>();
2870 const afterSequence = yield* orchestrator.getThreadEventSequence(threadId);
2871 yield* eventSink.stream({ threadId, afterSequence }).pipe(
2872 Stream.runForEach((stored) =>
2873 stored.event.type === "run.updated" && stored.event.payload.status === "starting"
2874 ? Queue.offer(startedRunIds, stored.event.payload.id)
2875 : Effect.void,
2876 ),
2877 Effect.forkScoped,
2878 );
2879 yield* Effect.yieldNow;
2880
2881 const now = yield* DateTime.now;
2882 // Runs start and settle the way the provider would report them.
2883 const settle = (key: string, startedAt: DateTime.Utc) =>
2884 Effect.gen(function* () {
2885 const run = yield* runFor(key);
2886 yield* eventSink.write({
2887 events: [
2888 {
2889 id: EventId.make(`runtime-layer-wake-work-start-${key}-completed`),
2890 type: "run.updated",
2891 threadId,
2892 runId: run.id,
2893 ...(run.rootNodeId === null ? {} : { nodeId: run.rootNodeId }),
2894 providerInstanceId: run.providerInstanceId,
2895 occurredAt: startedAt,
2896 payload: { ...run, status: "completed", startedAt, completedAt: startedAt },
2897 },
2898 ],
2899 });
2900 });
2901 const millis = (value: DateTime.Utc | undefined) =>
2902 value === undefined ? undefined : DateTime.toEpochMillis(value);
2903
2904 yield* settle("prompt", now);
2905 assert.equal(yield* Queue.take(startedRunIds), (yield* runFor("early-wake")).id);
2906 assert.equal(millis((yield* runFor("early-wake")).workStartedAt), millis(now));
2907
2908 yield* settle("early-wake", DateTime.add(now, { seconds: 1 }));
2909 assert.equal(yield* Queue.take(startedRunIds), (yield* runFor("queued")).id);
2910 assert.isUndefined((yield* runFor("queued")).workStartedAt);
2911
2912 // The queued prompt starts long after it was requested; the wake after it
2913 // counts from that start, not from the request.
2914 const queuedStartedAt = DateTime.add(now, { minutes: 10 });
2915 yield* settle("queued", queuedStartedAt);
2916 assert.equal(yield* Queue.take(startedRunIds), (yield* runFor("late-wake")).id);
2917 assert.equal(millis((yield* runFor("late-wake")).workStartedAt), millis(queuedStartedAt));
2918 }),
2919 );
2920
28032921 it.effect.each(["usage_limit", "provider_error"] as const)(
28042922 "handles a queued message after a %s failure",
28052923 (failureClass) =>
apps/web/src/components/Sidebar.logic.ts+3-4
......@@ -1147,10 +1147,9 @@ export function sortInboxThreadsByReturn<
11471147 );
11481148}
11491149
1150/** The timestamp a working thread's elapsed label counts from: the running
1151 turn's start (request time until adoption), falling back to the session's
1152 last transition when the turn projection lags behind. Malformed
1153 timestamps fall through to the next candidate, not just missing ones. */
1150/** The timestamp a working thread's elapsed label counts from: when its
1151 current work started (request time until adoption). Background wakes do
1152 not reset it. Malformed timestamps fall through to the next candidate. */
11541153export function resolveWorkingStartedAt(
11551154 thread: Pick<SidebarThreadSummary, "latestRun" | "runtime">,
11561155): string | null {
packages/client-runtime/src/state/models.ts+4-1
......@@ -296,7 +296,10 @@ export function resolveThreadProviderStack(
296296 return [...previous.slice(-(THREAD_PROVIDER_STACK_LIMIT - 1)), current];
297297}
298298
299/** Both shell and detail timers use the activity-owning run, never last activity. */
299/**
300 * Both shell and detail timers count from the activity-owning run's work
301 * start, never last activity. A wake keeps the start of the work it continues.
302 */
300303export function resolveThreadWorkingStartedAt(input: {
301304 readonly latestRun: Pick<
302305 ThreadRunSummary,
packages/client-runtime/src/state/threadExecution.test.ts+13
......@@ -96,6 +96,19 @@ describe("thread execution presentation", () => {
9696 ).toMatchObject({ status: "running", lastError: null, lastErrorClass: null });
9797 });
9898
99 it("counts a wake run's activity from the start of the work it continues", () => {
100 const workStartedAt = DateTime.makeUnsafe("2026-07-28T09:20:00.000Z");
101 const wake = { ...run("wake", 2, "running"), workStartedAt };
102 expect(
103 deriveThreadRuntime({ ...v2Projection, runs: [run("prompt", 1, "completed"), wake] })
104 ?.activityStartedAt,
105 ).toBe("2026-07-28T09:20:00.000Z");
106 expect(
107 deriveThreadRuntime({ ...v2Projection, runs: [run("prompt", 1, "running")] })
108 ?.activityStartedAt,
109 ).toBe("2026-07-28T10:00:00.000Z");
110 });
111
99112 it("keeps a subscription limit visible while later messages stay queued", () => {
100113 const failed = {
101114 ...run("limited", 1, "failed"),
packages/client-runtime/src/state/threadExecution.ts+7-3
......@@ -13,6 +13,7 @@ import {
1313 type ServerProviderModel,
1414 type OrchestrationV2ExecutionNode,
1515 type OrchestrationV2ThreadProjection,
16 orchestrationV2RunWorkStartedAt,
1617 type ThreadId,
1718} from "@t3tools/contracts";
1819import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork";
......@@ -213,6 +214,9 @@ export function deriveThreadRuntime(
213214 const usageLimitedRun = presentedUsageLimitRun(projection);
214215 const latestRunProjection = presentedLatestRun(projection);
215216 const activityRun = deriveThreadActivityRun(projection);
217 const liveActivityRun = latestMatchingRun(projection, (run) =>
218 ACTIVITY_RUN_STATUSES.has(run.status),
219 );
216220 if (latestRun === null && projection.thread.activeProviderThreadId === null) return null;
217221 const activeRunId =
218222 latestMatchingRun(projection, (run) => INTERRUPTIBLE_RUN_STATUSES.has(run.status))?.id ?? null;
......@@ -232,9 +236,9 @@ export function deriveThreadRuntime(
232236 : (activityRun?.status ?? "idle"),
233237 activeRunId,
234238 activityStartedAt:
235 activityRun !== null && ACTIVITY_RUN_STATUSES.has(activityRun.status)
236 ? (activityRun.startedAt ?? activityRun.requestedAt)
237 : null,
239 liveActivityRun === null
240 ? null
241 : DateTime.formatIso(orchestrationV2RunWorkStartedAt(liveActivityRun)),
238242 providerInstanceId: projection.thread.providerInstanceId,
239243 providerName: providerSession?.driver ?? null,
240244 ...threadErrorSummary(
packages/contracts/src/orchestrationV2.ts+21-1
......@@ -528,6 +528,12 @@ export const OrchestrationV2Run = Schema.Struct({
528528 contextHandoffId: Schema.NullOr(ContextHandoffId),
529529 /** Links server-generated restart continuations to the interrupted run. */
530530 restartContinuationOfRunId: Schema.optional(RunId),
531 /**
532 * Set on wake runs (background notifications, delegated task results,
533 * restart continuations): when the work they continue started. Read it
534 * through orchestrationV2RunWorkStartedAt.
535 */
536 workStartedAt: Schema.optional(Schema.DateTimeUtc),
531537 /**
532538 * Set by restart recovery on the thread's latest started run. Delivered to
533539 * the provider with the first later run that reaches a provider turn.
......@@ -545,6 +551,16 @@ export const OrchestrationV2Run = Schema.Struct({
545551});
546552export type OrchestrationV2Run = typeof OrchestrationV2Run.Type;
547553
554/**
555 * When the work a run belongs to started. A wake does not start new work, so
556 * working timers count from the prompt that did, not from the latest wake.
557 */
558export function orchestrationV2RunWorkStartedAt(
559 run: Pick<OrchestrationV2Run, "workStartedAt" | "startedAt" | "requestedAt">,
560): OrchestrationV2Run["requestedAt"] {
561 return run.workStartedAt ?? run.startedAt ?? run.requestedAt;
562}
563
548564export const OrchestrationV2RunAttempt = Schema.Struct({
549565 id: RunAttemptId,
550566 // Provider-thread rows can be reused after recovery; retain the native input destination.
......@@ -1688,7 +1704,10 @@ export const OrchestrationV2ThreadShell = Schema.Struct({
16881704 latestRunStartedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)),
16891705 latestRunCompletedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)),
16901706 activeRunId: Schema.NullOr(RunId),
1691 /** Start of the activity-owning run; request time while it is preparing. */
1707 /**
1708 * orchestrationV2RunWorkStartedAt of the activity-owning run: a wake keeps
1709 * the start of the work it continues; request time while preparing.
1710 */
16921711 activityRunStartedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)),
16931712 activityRunStatus: Schema.optional(
16941713 Schema.NullOr(Schema.Literals(["preparing", "starting", "running", "waiting"])),
......@@ -1841,6 +1860,7 @@ export const OrchestrationV2RunJson = OrchestrationV2Run.mapFields((fields) => (
18411860 requestedAt: Schema.DateTimeUtcFromString,
18421861 startedAt: Schema.NullOr(Schema.DateTimeUtcFromString),
18431862 completedAt: Schema.NullOr(Schema.DateTimeUtcFromString),
1863 workStartedAt: Schema.optional(Schema.DateTimeUtcFromString),
18441864}));
18451865export type OrchestrationV2RunJson = typeof OrchestrationV2RunJson.Type;
18461866