diff --git a/apps/server/src/orchestration-v2/Orchestrator.ts b/apps/server/src/orchestration-v2/Orchestrator.ts index 3157f55cacfa6..adb00b19119cd 100644 --- a/apps/server/src/orchestration-v2/Orchestrator.ts +++ b/apps/server/src/orchestration-v2/Orchestrator.ts @@ -2705,15 +2705,16 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio // the server or scheduled tasks, so a later completed run releases those unless it was stopped. const releaseUnattendedHolds = Effect.fn("Orchestrator.releaseUnattendedHolds")(function* ( threadId: ThreadId, - completedRunId: RunId, + completedRun: OrchestrationV2Run, ) { + const completedRunId = completedRun.id; const held = yield* projectionStore.getHeldQueuedRuns(threadId); if (held.length === 0) return; const projection = yield* projectionStore.getThreadRecords( threadId, ["runs", "messages", "turnItems"], { - messageIds: held.map((run) => run.userMessageId), + messageIds: [...held.map((run) => run.userMessageId), completedRun.userMessageId], turnItemTypes: ["run_interrupt_request"], turnItemRunIds: [completedRunId], }, @@ -2732,8 +2733,22 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio .filter((message) => message.createdBy !== "user" || message.scheduledTaskId !== undefined) .map((message) => message.id), ); + // A restart that cut a compaction records the holds it added; redoing it releases those. + // Retried or re-cut continuations chain back to the run that recorded them. + const restartHeld = new Set(); + const isCompaction = messages.some( + (message) => message.id === completedRun.userMessageId && isCompactionCommandMessage(message), + ); + let id = isCompaction ? completedRun.restartContinuationOfRunId : undefined; + for (let hops = 0; id !== undefined && hops < projection.runs.length; hops++) { + const source = projection.runs.find((run) => run.id === id); + for (const heldId of source?.restartHeldRunIds ?? []) restartHeld.add(heldId); + id = source?.restartContinuationOfRunId; + } const now = yield* DateTime.now; - const released = held.filter((run) => unattended.has(run.userMessageId)); + const released = held.filter( + (run) => unattended.has(run.userMessageId) || restartHeld.has(run.id), + ); if (released.length === 0) return; yield* writeSystemEvents( released.map((run) => ({ @@ -14237,10 +14252,7 @@ const makeOrchestrator = Effect.fn("orchestrationV2.Orchestrator.layer")(functio yield* threadDispatch .withLock( threadId, - (run.status === "completed" - ? releaseUnattendedHolds(threadId, run.id) - : Effect.void - ).pipe( + (run.status === "completed" ? releaseUnattendedHolds(threadId, run) : Effect.void).pipe( Effect.andThen( startNextQueuedRun(threadId, { endedRunId: run.id, diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts index 642a642a80df1..1066f81578dea 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.test.ts @@ -710,6 +710,8 @@ it.effect( providerThreadId, providerInstanceId: ProviderInstanceId.make("codex"), }, + { id: RunId.make("run_recovery_queued"), status: "queued", ordinal: 2 }, + { id: RunId.make("run_recovery_held"), status: "queued", ordinal: 3, queueHeld: true }, ], attempts: [ { @@ -746,6 +748,8 @@ it.effect( Layer.mock(ProjectionStore.ProjectionStoreV2)({ getRecoveryThreadIds: () => Effect.succeed([threadId]), getRuntimeRecoveryProjection: () => Effect.succeed(projection), + getMessagesById: () => Effect.succeed(new Map()), + getThreadRecords: () => Effect.succeed(projection), }), Layer.mock(EventSink.EventSinkV2)({ commitCommand: (input) => { @@ -784,6 +788,7 @@ it.effect( "status" in event.payload ? event.payload.status : null, ]), [ + ["run.updated", "queued"], ["run.updated", "cancelled"], ["run-attempt.updated", "cancelled"], ["node.updated", "cancelled"], @@ -797,6 +802,13 @@ it.effect( ["provider-session.updated", "stopped"], ], ); + // Only the hold this restart added is recorded on the run it cut. + const cutRun = events.find( + (event) => event.type === "run.updated" && event.payload.status === "cancelled", + ); + assert.deepEqual(cutRun?.type === "run.updated" ? cutRun.payload.restartHeldRunIds : null, [ + RunId.make("run_recovery_queued"), + ]); const messageEvent = events.find((event) => event.type === "message.updated"); assert.isFalse(messageEvent?.type === "message.updated" && messageEvent.payload.streaming); const stoppedSession = events.find((event) => event.type === "provider-session.updated"); diff --git a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts index 89e43e5b1ede4..c32d1c5b7aa19 100644 --- a/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts +++ b/apps/server/src/orchestration-v2/ProviderRuntimeRecoveryService.ts @@ -5,6 +5,7 @@ import { isProviderNativeSubagentThread, type OrchestrationV2DomainEvent, type ProviderThreadId, + type RunId, type OrchestrationV2RestartCancelledBackgroundWork, type OrchestrationV2Subagent, type OrchestrationV2ThreadProjection, @@ -421,12 +422,14 @@ export const make = Effect.gen(function* () { ...retryableRuns, ...queuedRunsInDeliveryOrder({ runs: queuedRuns, messages: queuedMessages }), ]; + const restartHeldRunIds: Array = []; for (const [index, run] of heldRuns.entries()) { if (bufferedMessageIds.has(run.userMessageId)) { runs.push(run); continue; } if (run.queueHeld === true && retryableRuns.length === 0) continue; + if (run.queueHeld !== true) restartHeldRunIds.push(run.id); events.push({ id: yield* allocateEventId(), type: "run.updated", @@ -536,7 +539,13 @@ export const make = Effect.gen(function* () { runId: run.id, providerInstanceId: run.providerInstanceId, occurredAt: now, - payload: { ...run, status: "cancelled", queuePosition: null, completedAt: now }, + payload: { + ...run, + status: "cancelled", + queuePosition: null, + completedAt: now, + ...(restartHeldRunIds.length > 0 ? { restartHeldRunIds } : {}), + }, }); for (const attempt of projection.attempts.filter( (candidate) => diff --git a/apps/server/src/orchestration-v2/RestartContinuation.test.ts b/apps/server/src/orchestration-v2/RestartContinuation.test.ts index d2f21a772ac99..6df1b633a3845 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.test.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.test.ts @@ -1077,14 +1077,29 @@ it.effect("does not continue a run the user asked to stop before the restart", ( }), ); -it.effect("does not continue a cut /compact or /logout turn", () => +it.effect("compacts again after a cut /compact turn, and does not continue a cut /logout", () => Effect.gen(function* () { - for (const text of ["/compact", " /LOGOUT "]) { - const texts = yield* continuationTexts({ + for (const [text, creationSource, expected] of [ + [" /Compact ", "mcp", ["user:/compact"]], + // Servers before `idleCompaction` marked their own compaction this way. + ["/compact", "server", ["agent:/compact"]], + [" /LOGOUT ", "mcp", []], + ] as const) { + const sent = yield* continuationDispatches({ ...cutMidTurn(), - messages: [{ id: MessageId.make("message:user"), text, attachments: [] }], + messages: [ + { + id: MessageId.make("message:user"), + role: "user", + createdBy: "agent", + creationSource, + text, + attachments: [], + }, + ], } as unknown as OrchestrationV2ThreadProjection); - assert.deepEqual(texts, [], text); + const texts = sent.map((command) => `${command.createdBy}:${command.text}`); + assert.deepEqual(texts, [...expected], text); } const texts = yield* continuationTexts({ ...cutMidTurn(), diff --git a/apps/server/src/orchestration-v2/RestartContinuation.ts b/apps/server/src/orchestration-v2/RestartContinuation.ts index 20d110d335aa2..330927f96fdad 100644 --- a/apps/server/src/orchestration-v2/RestartContinuation.ts +++ b/apps/server/src/orchestration-v2/RestartContinuation.ts @@ -2,6 +2,9 @@ import { runRanAfter } from "@t3tools/shared/orchestrationV2ThreadError"; import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { CommandId, + IDLE_COMPACTION_TEXT, + idleCompactionOf, + isCompactionCommandMessage, isOrchestrationV2WorkActive, latestProviderTurnForAttempt, MessageId, @@ -290,7 +293,13 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta const sourceMessage = sourceRecords.messages.find( (message) => message.id === source.userMessageId, ); - if (sourceMessage !== undefined && isNativeMaintenanceCommand(sourceMessage)) return; + // Compaction changes nothing a repeat could double, so a cut one runs again. + const recompact = + sourceMessage !== undefined && + isCompactionCommandMessage(sourceMessage) && + !isRestartNoteSource(source, projection.providerTurns); + if (!recompact && sourceMessage !== undefined && isNativeMaintenanceCommand(sourceMessage)) + return; const note = restartContinuationNote( source, projection.runs, @@ -311,19 +320,27 @@ export const continueRestartedRun = Effect.fn("RestartContinuation.continueResta commandId: CommandId.make(`command:restart-continuation:${input.sourceRunId}`), threadId: input.threadId, messageId, - text, - notification: { - source: { kind: "system" }, - outcome: "updated", - summary: note.settled - ? "T3 Code restarted and stopped background work" - : "T3 Code restarted and resumed this turn", - detail: text, - }, + text: recompact ? IDLE_COMPACTION_TEXT : text, + idleCompaction: recompact ? sourceMessage.idleCompaction : undefined, + notification: recompact + ? undefined + : { + source: { kind: "system" }, + outcome: "updated", + summary: note.settled + ? "T3 Code restarted and stopped background work" + : "T3 Code restarted and resumed this turn", + detail: text, + }, attachments: [], modelSelection: source.modelSelection, dispatchMode: { type: "start_immediately" }, - createdBy: "agent", + // Agent + server with no idleCompaction reads as an automatic idle compaction. + createdBy: !recompact + ? "agent" + : idleCompactionOf(sourceMessage) === undefined + ? "user" + : sourceMessage.createdBy, creationSource: "server", restartContinuationOfRunId: input.sourceRunId, }) diff --git a/apps/server/src/orchestration-v2/runtimeLayer.test.ts b/apps/server/src/orchestration-v2/runtimeLayer.test.ts index a49e323770cc1..caf95cf9445f8 100644 --- a/apps/server/src/orchestration-v2/runtimeLayer.test.ts +++ b/apps/server/src/orchestration-v2/runtimeLayer.test.ts @@ -5597,7 +5597,7 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }); const end = ( key: string, - status: "completed" | "failed", + status: "completed" | "failed" | "cancelled", revision = "", patch: Partial = {}, ) => @@ -5747,6 +5747,31 @@ it.layer(layerTest)("OrchestrationV2LayerLive lifecycle", (it) => { }), ); + it.effect("a compaction resumed after a restart releases only what that restart held", () => + Effect.gen(function* () { + const threadId = ThreadId.make("runtime-layer-held-queue-resumed-compaction"); + const { send, runFor, end, startedRunIds, awaitHeld } = yield* queueTestThread(threadId); + yield* send("failed", "agent", true); + yield* send("earlier", "user", false); + yield* send("later", "user", false); + yield* end("failed", "failed"); + yield* awaitHeld("later"); + yield* send("cut", "agent", true); + yield* end("cut", "cancelled", "", { restartHeldRunIds: [(yield* runFor("later")).id] }); + // A second restart cuts the first continuation, which adds no holds of its own. + yield* send(" /compact", "agent", true); + yield* end(" /compact", "cancelled", "", { + restartContinuationOfRunId: (yield* runFor("cut")).id, + }); + yield* send("/compact", "agent", true); + yield* end("/compact", "completed", "", { + restartContinuationOfRunId: (yield* runFor(" /compact")).id, + }); + assert.equal(yield* Queue.take(startedRunIds), (yield* runFor("later")).id); + assert.isTrue((yield* runFor("earlier")).queueHeld); + }), + ); + it.effect("keeps held agent messages while the thread's worktree setup is unresolved", () => Effect.gen(function* () { const orchestrator = yield* Orchestrator.OrchestratorV2; diff --git a/docs/internals/server-updates.md b/docs/internals/server-updates.md index 1536884578d4b..6ec9ee5daca5d 100644 --- a/docs/internals/server-updates.md +++ b/docs/internals/server-updates.md @@ -87,8 +87,9 @@ cannot be overwritten by a stale cancellation. The [continuation handler](../../apps/server/src/orchestration-v2/RestartContinuation.ts) rechecks the preference, provider selection, newer user work, a stop the user requested, -and maintenance turns such as `/compact` before dispatching. Intent and delivery both -refuse a thread that is archived, settled, or still snoozed, a cut turn whose provider +and maintenance turns such as `/logout` before dispatching. A cut `/compact` runs again, and +its completion releases the holds that restart added, since they waited on it, not a person. +Intent and delivery both refuse a thread that is archived, settled, or still snoozed, a cut turn whose provider turn already ended, and a run waiting on the user: an approval, question, or secret that was pending, or that recovery closed on the cut turn. Message-capable questions stay answerable, so they do not block. Delivery also stops after three consecutive continuations in one diff --git a/docs/user/composer.md b/docs/user/composer.md index df880e49c6ca5..95c35b2e4abc5 100644 --- a/docs/user/composer.md +++ b/docs/user/composer.md @@ -261,6 +261,7 @@ If you stop the agent or the server restarts, queued messages keep their order a Press **Resume** in an empty composer to continue. **Resume** also continues a stopped run or one that hit a usage limit, even with nothing queued. Sending a new message also resumes the queue, unless it steers a turn that is already running. You can edit or remove held messages without starting them. +A `/compact` cut by a restart runs again, and messages queued behind it send when it finishes. The pencil on a queued message opens it in the composer for editing. The original message stays in the queue until you save, and it is highlighted while you edit. The message's diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts index b1fd02b3d3a07..630c2b7517226 100644 --- a/packages/contracts/src/orchestrationV2.ts +++ b/packages/contracts/src/orchestrationV2.ts @@ -706,6 +706,8 @@ export const OrchestrationV2Run = Schema.Struct({ restartCancelledBackgroundWork: Schema.optional( Schema.Array(OrchestrationV2RestartCancelledBackgroundWork), ), + /** Queued runs a restart held when it cut this run, as opposed to holds already in place. */ + restartHeldRunIds: Schema.optional(Schema.Array(RunId)), sourcePlanRef: Schema.optional( Schema.Struct({ threadId: ThreadId,