diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index c58dd46609ec3bd972e0cd9cdbfad4edf81f7eaf..c52087778d63d6c2bfb849342139aeac72418527 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -39,6 +39,7 @@ import { type OrchestrationV2Subagent, type OrchestrationV2ThreadProjection, type OrchestrationV2TurnItem, + orchestrationV2RunWorkStartedAt, ProviderInstanceId, type ProviderSessionId, RunId, @@ -304,6 +305,37 @@ function nextRunOrdinal(projection: Pick, + trigger: { + readonly notification?: unknown; + readonly delegatedCompletion?: unknown; + readonly restartContinuationOfRunId?: RunId | undefined; + }, +): Pick { + if ( + trigger.notification === undefined && + trigger.delegatedCompletion === undefined && + trigger.restartContinuationOfRunId === undefined + ) { + return {}; + } + const previous = runs + .flatMap((run) => (run.startedAt === null ? [] : [{ run, startedAt: run.startedAt }])) + .toSorted( + (left, right) => + DateTime.Order(right.startedAt, left.startedAt) || right.run.ordinal - left.run.ordinal, + )[0]?.run; + return previous === undefined ? {} : { workStartedAt: orchestrationV2RunWorkStartedAt(previous) }; +} + function isNativeMaintenanceCommand(message: { readonly text: string; readonly attachments: ReadonlyArray; @@ -1459,6 +1491,11 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio queuePosition: null, startedAt: null, contextHandoffId: activeHandoff?.id ?? null, + ...wakeWorkStartedAt(projection.runs, { + notification: queuedMessage.notification, + delegatedCompletion: queuedMessage.delegatedCompletion, + restartContinuationOfRunId: queuedRun.restartContinuationOfRunId, + }), }; const userTurnItem: OrchestrationV2TurnItem = { ...(legacyQueuedTurnItem ?? { @@ -4949,6 +4986,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(command.restartContinuationOfRunId === undefined ? {} : { restartContinuationOfRunId: command.restartContinuationOfRunId }), + ...wakeWorkStartedAt(projection.runs, command), }; const attempt: OrchestrationV2RunAttempt = { id: attemptId, @@ -5642,6 +5680,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio ...(command.restartContinuationOfRunId === undefined ? {} : { restartContinuationOfRunId: command.restartContinuationOfRunId }), + ...wakeWorkStartedAt(projection.runs, command), }; const attempt: OrchestrationV2RunAttempt = { id: attemptId, diff --git a/apps/server/src/orchestration-v2/ProjectionStore.test.ts b/apps/server/src/orchestration-v2/ProjectionStore.test.ts index f3b6348c5d31c44d1f2d1cd993de33c65921038d..c64fece9db0700a0e6f71e70bc226babcad26a0f 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.test.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.test.ts @@ -1982,6 +1982,37 @@ it.layer(TestLayer)("ProjectionStoreV2", (it) => { assert.isNull(onlyHeldShell.latestRunId); assert.equal(onlyHeldShell.status, "idle"); } + + // A wake run counts from the start of the work it continues. + yield* projectionStore.apply({ + id: EventId.make("event:projection-shell-interruptible:wake"), + type: "run.updated", + threadId, + runId, + nodeId: rootNodeId, + driver, + occurredAt: later, + payload: { + ...run, + status: "running", + requestedAt: later, + startedAt: later, + workStartedAt: now, + }, + }); + const wakeProjection = yield* projectionStore.getThreadProjection(threadId); + const wakeSqlShell = (yield* projectionStore.getShellSnapshot()).threads.find( + (row) => row.id === threadId, + )!; + for (const wakeShell of [ + wakeSqlShell, + ProjectionStore.threadShellFromProjection(wakeProjection), + ]) { + assert.equal( + wakeShell.activityRunStartedAt && DateTime.toEpochMillis(wakeShell.activityRunStartedAt), + DateTime.toEpochMillis(now), + ); + } }), ); diff --git a/apps/server/src/orchestration-v2/ProjectionStore.ts b/apps/server/src/orchestration-v2/ProjectionStore.ts index 6718240d9e7aa223820c14d131a764bb3aca4ee8..b662a528b4dea5ab85d9af5af3bef90e569e8d16 100644 --- a/apps/server/src/orchestration-v2/ProjectionStore.ts +++ b/apps/server/src/orchestration-v2/ProjectionStore.ts @@ -48,6 +48,7 @@ import { OrchestrationV2RuntimeRequestJson as OrchestrationV2RuntimeRequestJsonSchema, OrchestrationV2SubagentJson as OrchestrationV2SubagentJsonSchema, OrchestrationV2TurnItemJson as OrchestrationV2TurnItemJsonSchema, + orchestrationV2RunWorkStartedAt, RunId, CheckpointScopeId, ThreadId, @@ -1368,7 +1369,8 @@ export function threadShellFromProjection( latestRunCompletedAt: latestRun?.completedAt ?? null, activeRunId: activeRun?.id ?? null, activityRunStatus: activityRun?.status ?? null, - activityRunStartedAt: activityRun?.startedAt ?? activityRun?.requestedAt ?? null, + activityRunStartedAt: + activityRun === null ? null : orchestrationV2RunWorkStartedAt(activityRun), status: latestRun?.status ?? "idle", ...threadErrorSummary( latestRootProviderFailure(latestRun, projection.turnItems), @@ -4782,7 +4784,12 @@ export const layer: Layer.Layer = LIMIT 1 ) AS activity_run_status, ( - SELECT COALESCE(json_extract(r.payload_json, '$.startedAt'), r.requested_at) + -- Mirrors orchestrationV2RunWorkStartedAt. + SELECT COALESCE( + json_extract(r.payload_json, '$.workStartedAt'), + json_extract(r.payload_json, '$.startedAt'), + r.requested_at + ) FROM orchestration_v2_projection_runs r WHERE r.thread_id = t.thread_id AND r.status IN ('preparing', 'starting', 'running', 'waiting') diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index 7f66cebd27d1c7ba8f1b9556e4b5adfa85f8a641..d077f7e209107a1d2d65aa7a386c9b931cbf37c5 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -2800,6 +2800,124 @@ it.layer(TestLayer)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("starts a wake's work clock from the run that ran before it", () => + Effect.gen(function* () { + const orchestrator = yield* Orchestrator.OrchestratorV2; + const eventSink = yield* EventSink.EventSinkV2; + const threadId = ThreadId.make("runtime-layer-wake-work-start-thread"); + const messageId = (key: string) => MessageId.make(`runtime-layer-wake-work-start-${key}`); + + yield* orchestrator.dispatch({ + type: "thread.create", + createdBy: "user", + creationSource: "web", + commandId: CommandId.make("runtime-layer-wake-work-start-create"), + threadId, + projectId: ProjectId.make("runtime-layer-wake-work-start-project"), + title: "Wake work start", + modelSelection, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: process.cwd(), + }); + const dispatch = (key: string, wake: boolean) => + orchestrator.dispatch({ + type: "message.dispatch", + createdBy: wake ? "agent" : "user", + creationSource: wake ? "provider" : "web", + ...(wake + ? { + notification: { + source: { kind: "background_task" as const }, + outcome: "updated" as const, + summary: "Background activity updated", + }, + } + : {}), + commandId: CommandId.make(`runtime-layer-wake-work-start-${key}`), + threadId, + messageId: messageId(key), + text: key, + attachments: [], + modelSelection, + dispatchMode: + key === "prompt" ? { type: "start_immediately" } : { type: "queue_after_active" }, + }); + const runFor = (key: string) => + Effect.map(orchestrator.getThreadProjection(threadId), ({ runs }) => { + const run = runs.find((candidate) => candidate.userMessageId === messageId(key)); + assert.isDefined(run); + return run; + }); + yield* dispatch("prompt", false); + yield* dispatch("queued", false); + yield* dispatch("early-wake", true); + // The early wake now runs ahead of the older queued prompt, as a + // delegated result does when it jumps the queue. + yield* orchestrator.dispatch({ + type: "queued-run.reorder", + commandId: CommandId.make("runtime-layer-wake-work-start-reorder"), + threadId, + runId: (yield* runFor("queued")).id, + beforeRunId: null, + }); + yield* dispatch("late-wake", true); + // A queued wake has no clock yet: what runs before it is still unknown. + assert.isUndefined((yield* runFor("early-wake")).workStartedAt); + + const startedRunIds = yield* Queue.unbounded(); + const afterSequence = yield* orchestrator.getThreadEventSequence(threadId); + yield* eventSink.stream({ threadId, afterSequence }).pipe( + Stream.runForEach((stored) => + stored.event.type === "run.updated" && stored.event.payload.status === "starting" + ? Queue.offer(startedRunIds, stored.event.payload.id) + : Effect.void, + ), + Effect.forkScoped, + ); + yield* Effect.yieldNow; + + const now = yield* DateTime.now; + // Runs start and settle the way the provider would report them. + const settle = (key: string, startedAt: DateTime.Utc) => + Effect.gen(function* () { + const run = yield* runFor(key); + yield* eventSink.write({ + events: [ + { + id: EventId.make(`runtime-layer-wake-work-start-${key}-completed`), + type: "run.updated", + threadId, + runId: run.id, + ...(run.rootNodeId === null ? {} : { nodeId: run.rootNodeId }), + providerInstanceId: run.providerInstanceId, + occurredAt: startedAt, + payload: { ...run, status: "completed", startedAt, completedAt: startedAt }, + }, + ], + }); + }); + const millis = (value: DateTime.Utc | undefined) => + value === undefined ? undefined : DateTime.toEpochMillis(value); + + yield* settle("prompt", now); + assert.equal(yield* Queue.take(startedRunIds), (yield* runFor("early-wake")).id); + assert.equal(millis((yield* runFor("early-wake")).workStartedAt), millis(now)); + + yield* settle("early-wake", DateTime.add(now, { seconds: 1 })); + assert.equal(yield* Queue.take(startedRunIds), (yield* runFor("queued")).id); + assert.isUndefined((yield* runFor("queued")).workStartedAt); + + // The queued prompt starts long after it was requested; the wake after it + // counts from that start, not from the request. + const queuedStartedAt = DateTime.add(now, { minutes: 10 }); + yield* settle("queued", queuedStartedAt); + assert.equal(yield* Queue.take(startedRunIds), (yield* runFor("late-wake")).id); + assert.equal(millis((yield* runFor("late-wake")).workStartedAt), millis(queuedStartedAt)); + }), + ); + it.effect.each(["usage_limit", "provider_error"] as const)( "handles a queued message after a %s failure", (failureClass) => diff --git a/apps/web/src/components/Sidebar.logic.ts b/apps/web/src/components/Sidebar.logic.ts index ad2c9a1fedc8dfb5ed770916bcd1748a608d7b7c..4f504e286d0c36d90d68954f620df29643176068 100644 --- a/apps/web/src/components/Sidebar.logic.ts +++ b/apps/web/src/components/Sidebar.logic.ts @@ -1147,10 +1147,9 @@ export function sortInboxThreadsByReturn< ); } -/** The timestamp a working thread's elapsed label counts from: the running - turn's start (request time until adoption), falling back to the session's - last transition when the turn projection lags behind. Malformed - timestamps fall through to the next candidate, not just missing ones. */ +/** The timestamp a working thread's elapsed label counts from: when its + current work started (request time until adoption). Background wakes do + not reset it. Malformed timestamps fall through to the next candidate. */ export function resolveWorkingStartedAt( thread: Pick, ): string | null { diff --git a/packages/client-runtime/src/state/models.ts b/packages/client-runtime/src/state/models.ts index 7163bd79a8205dbace244a56b0e129838f287f78..37b7b5b8e3036b41ed6d45d1b0f03365b65cddd8 100644 --- a/packages/client-runtime/src/state/models.ts +++ b/packages/client-runtime/src/state/models.ts @@ -296,7 +296,10 @@ export function resolveThreadProviderStack( return [...previous.slice(-(THREAD_PROVIDER_STACK_LIMIT - 1)), current]; } -/** Both shell and detail timers use the activity-owning run, never last activity. */ +/** + * Both shell and detail timers count from the activity-owning run's work + * start, never last activity. A wake keeps the start of the work it continues. + */ export function resolveThreadWorkingStartedAt(input: { readonly latestRun: Pick< ThreadRunSummary, diff --git a/packages/client-runtime/src/state/threadExecution.test.ts b/packages/client-runtime/src/state/threadExecution.test.ts index 7ebd468400f8793b1ba5bdbb72d6771fd440887e..7a94233e34e95b165fa00193a6fbb77be9161931 100644 --- a/packages/client-runtime/src/state/threadExecution.test.ts +++ b/packages/client-runtime/src/state/threadExecution.test.ts @@ -96,6 +96,19 @@ describe("thread execution presentation", () => { ).toMatchObject({ status: "running", lastError: null, lastErrorClass: null }); }); + it("counts a wake run's activity from the start of the work it continues", () => { + const workStartedAt = DateTime.makeUnsafe("2026-07-28T09:20:00.000Z"); + const wake = { ...run("wake", 2, "running"), workStartedAt }; + expect( + deriveThreadRuntime({ ...v2Projection, runs: [run("prompt", 1, "completed"), wake] }) + ?.activityStartedAt, + ).toBe("2026-07-28T09:20:00.000Z"); + expect( + deriveThreadRuntime({ ...v2Projection, runs: [run("prompt", 1, "running")] }) + ?.activityStartedAt, + ).toBe("2026-07-28T10:00:00.000Z"); + }); + it("keeps a subscription limit visible while later messages stay queued", () => { const failed = { ...run("limited", 1, "failed"), diff --git a/packages/client-runtime/src/state/threadExecution.ts b/packages/client-runtime/src/state/threadExecution.ts index 5101acc062ef1f0a5ca0afc538122c0d72b77bbf..707295891119add400c4c29a132f67e0a57eba44 100644 --- a/packages/client-runtime/src/state/threadExecution.ts +++ b/packages/client-runtime/src/state/threadExecution.ts @@ -13,6 +13,7 @@ import { type ServerProviderModel, type OrchestrationV2ExecutionNode, type OrchestrationV2ThreadProjection, + orchestrationV2RunWorkStartedAt, type ThreadId, } from "@t3tools/contracts"; import { derivePendingBackgroundWork } from "@t3tools/shared/orchestrationV2PendingBackgroundWork"; @@ -213,6 +214,9 @@ export function deriveThreadRuntime( const usageLimitedRun = presentedUsageLimitRun(projection); const latestRunProjection = presentedLatestRun(projection); const activityRun = deriveThreadActivityRun(projection); + const liveActivityRun = latestMatchingRun(projection, (run) => + ACTIVITY_RUN_STATUSES.has(run.status), + ); if (latestRun === null && projection.thread.activeProviderThreadId === null) return null; const activeRunId = latestMatchingRun(projection, (run) => INTERRUPTIBLE_RUN_STATUSES.has(run.status))?.id ?? null; @@ -232,9 +236,9 @@ export function deriveThreadRuntime( : (activityRun?.status ?? "idle"), activeRunId, activityStartedAt: - activityRun !== null && ACTIVITY_RUN_STATUSES.has(activityRun.status) - ? (activityRun.startedAt ?? activityRun.requestedAt) - : null, + liveActivityRun === null + ? null + : DateTime.formatIso(orchestrationV2RunWorkStartedAt(liveActivityRun)), providerInstanceId: projection.thread.providerInstanceId, providerName: providerSession?.driver ?? null, ...threadErrorSummary( diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index cb220a5b4ccd67da59b338477bfa3fa285512ae2..9d907c81785dd294c35e2e60f9e91b8b85e19384 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -528,6 +528,12 @@ export const OrchestrationV2Run = Schema.Struct({ contextHandoffId: Schema.NullOr(ContextHandoffId), /** Links server-generated restart continuations to the interrupted run. */ restartContinuationOfRunId: Schema.optional(RunId), + /** + * Set on wake runs (background notifications, delegated task results, + * restart continuations): when the work they continue started. Read it + * through orchestrationV2RunWorkStartedAt. + */ + workStartedAt: Schema.optional(Schema.DateTimeUtc), /** * Set by restart recovery on the thread's latest started run. Delivered to * the provider with the first later run that reaches a provider turn. @@ -545,6 +551,16 @@ export const OrchestrationV2Run = Schema.Struct({ }); export type OrchestrationV2Run = typeof OrchestrationV2Run.Type; +/** + * When the work a run belongs to started. A wake does not start new work, so + * working timers count from the prompt that did, not from the latest wake. + */ +export function orchestrationV2RunWorkStartedAt( + run: Pick, +): OrchestrationV2Run["requestedAt"] { + return run.workStartedAt ?? run.startedAt ?? run.requestedAt; +} + export const OrchestrationV2RunAttempt = Schema.Struct({ id: RunAttemptId, // Provider-thread rows can be reused after recovery; retain the native input destination. @@ -1688,7 +1704,10 @@ export const OrchestrationV2ThreadShell = Schema.Struct({ latestRunStartedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)), latestRunCompletedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)), activeRunId: Schema.NullOr(RunId), - /** Start of the activity-owning run; request time while it is preparing. */ + /** + * orchestrationV2RunWorkStartedAt of the activity-owning run: a wake keeps + * the start of the work it continues; request time while preparing. + */ activityRunStartedAt: Schema.optional(Schema.NullOr(Schema.DateTimeUtc)), activityRunStatus: Schema.optional( Schema.NullOr(Schema.Literals(["preparing", "starting", "running", "waiting"])), @@ -1841,6 +1860,7 @@ export const OrchestrationV2RunJson = OrchestrationV2Run.mapFields((fields) => ( requestedAt: Schema.DateTimeUtcFromString, startedAt: Schema.NullOr(Schema.DateTimeUtcFromString), completedAt: Schema.NullOr(Schema.DateTimeUtcFromString), + workStartedAt: Schema.optional(Schema.DateTimeUtcFromString), })); export type OrchestrationV2RunJson = typeof OrchestrationV2RunJson.Type;