diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts index 00daff5a8c87..6d0243e5e5f9 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.test.ts @@ -6,7 +6,9 @@ import { EventId, ProviderInstanceId, ProviderSessionId, + RunId, ThreadId, + type MessageId, type OrchestrationProjectShell, type OrchestrationV2AppThread, type OrchestrationV2Command, @@ -527,6 +529,8 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: const settingsChanges = yield* PubSub.unbounded(); const mergedPullRequests = yield* PubSub.unbounded(); const commands = yield* Ref.make>([]); + /** Commands the settle prompt sends: the prompt, an interrupt, the settle after it. */ + const promptCommands = yield* Queue.unbounded(); const branchCalls = yield* Ref.make< ReadonlyArray<{ readonly cwd: string; readonly branch: string }> >([]); @@ -578,6 +582,15 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: }); const dispatch: Orchestrator.OrchestratorV2Shape["dispatch"] = (command) => { + if ( + command.type === "message.dispatch" || + command.type === "thread.settle" || + command.type === "run.interrupt" + ) { + return Queue.offer(promptCommands, command).pipe( + Effect.as({ sequence: 1, storedEvents: [] }), + ); + } if (command.type !== "thread.auto-settle") { return Effect.die(new Error(`Unexpected command: ${command.type}`)); } @@ -669,6 +682,7 @@ const makeHarness = Effect.fn("makeThreadSettlementHarness")(function* (options: invalidatedCwds, closedIdle, scriptRuns, + promptCommands, publishEvent: (event: OrchestrationV2DomainEvent) => PubSub.publish(domainEvents, event), updateSettings, publishMerge: PubSub.publish(mergedPullRequests, { @@ -1236,6 +1250,143 @@ describe("ThreadSettlementServiceV2 terminals", () => { }), ), ); + + describe("settle prompt", () => { + const settings = { ...DEFAULT_SERVER_SETTINGS, settlePrompt: "/wrap-up" }; + const runEvent = ( + thread: OrchestrationV2AppThread, + userMessageId: MessageId, + status: "running" | "completed", + ): OrchestrationV2DomainEvent => ({ + type: "run.updated", + id: EventId.make(`event:run:${thread.id}:${status}`), + threadId: thread.id, + occurredAt: DateTime.makeUnsafe(NOW), + payload: { + id: RunId.make(`run:${thread.id}`), + threadId: thread.id, + ordinal: 2, + providerInstanceId: ProviderInstanceId.make("codex"), + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5" }, + providerThreadId: null, + userMessageId, + rootNodeId: null, + activeAttemptId: null, + status, + requestedAt: DateTime.makeUnsafe(NOW), + startedAt: null, + completedAt: null, + checkpointId: null, + contextHandoffId: null, + }, + }); + const takePrompt = (queue: Queue.Queue) => + Queue.take(queue).pipe( + Effect.map((command) => { + assert.strictEqual(command.type, "message.dispatch"); + return command as Extract; + }), + ); + + it.effect("sends the prompt, settles again when its turn ends, then cleans up", () => + Effect.scoped( + Effect.gen(function* () { + const thread = appThread("wrap-up-thread", "settled"); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([]), + settings, + currentThreads: [thread], + }); + + yield* Effect.gen(function* () { + const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; + yield* startHarness(service, fixture.activation, fixture.snapshotReads); + yield* fixture.publishEvent(settledEvent(thread)); + const prompt = yield* takePrompt(fixture.promptCommands); + assert.strictEqual(prompt.threadId, thread.id); + assert.strictEqual(prompt.text, "/wrap-up"); + assert.strictEqual(prompt.createdBy, "system"); + assert.deepStrictEqual(prompt.dispatchMode, { type: "start_immediately" }); + + // Its turn ends: the thread settles again. Cleanup waits for that settle. + yield* fixture.publishEvent(runEvent(thread, prompt.messageId, "running")); + yield* fixture.publishEvent(runEvent(thread, prompt.messageId, "completed")); + const settle = yield* Queue.take(fixture.promptCommands); + assert.deepStrictEqual( + { type: settle.type, threadId: "threadId" in settle ? settle.threadId : null }, + { type: "thread.settle", threadId: thread.id }, + ); + assert.strictEqual(yield* Queue.size(fixture.closedIdle), 0); + + // The settle that follows the prompt cleans up and sends nothing more. + const resettled = { + ...thread, + settledAt: DateTime.makeUnsafe(Date.parse(NOW) + 60_000), + }; + yield* fixture.publishEvent(settledEvent(resettled)); + assert.deepStrictEqual(yield* Queue.take(fixture.closedIdle), { threadId: thread.id }); + assert.strictEqual(yield* Queue.size(fixture.promptCommands), 0); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("sends the prompt once when a settlement is re-emitted", () => + Effect.scoped( + Effect.gen(function* () { + const thread = appThread("re-emitted-thread", "settled"); + const marker = appThread("marker-thread", "settled"); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([]), + settings: { ...DEFAULT_SERVER_SETTINGS }, + currentThreads: [thread, marker], + }); + + yield* Effect.gen(function* () { + const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; + yield* startHarness(service, fixture.activation, fixture.snapshotReads); + yield* fixture.updateSettings({ settlePrompt: "/wrap-up" }); + yield* fixture.publishEvent(settledEvent(thread)); + yield* takePrompt(fixture.promptCommands); + yield* fixture.publishEvent(settledEvent(thread)); + // Events run in order: once the marker is prompted, the re-emitted + // settlement has been handled without a second prompt. + yield* fixture.publishEvent(settledEvent(marker)); + const next = yield* takePrompt(fixture.promptCommands); + assert.strictEqual(next.threadId, marker.id); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + + it.effect("stops a prompt turn that runs past the timeout, then settles", () => + Effect.scoped( + Effect.gen(function* () { + const thread = appThread("slow-thread", "settled"); + const fixture = yield* makeHarness({ + snapshot: makeSnapshot([]), + settings, + currentThreads: [thread], + }); + + yield* Effect.gen(function* () { + const service = yield* ThreadSettlementService.ThreadSettlementServiceV2; + yield* startHarness(service, fixture.activation, fixture.snapshotReads); + yield* fixture.publishEvent(settledEvent(thread)); + const prompt = yield* takePrompt(fixture.promptCommands); + yield* fixture.publishEvent(runEvent(thread, prompt.messageId, "running")); + yield* TestClock.adjust(ThreadSettlementService.SETTLE_PROMPT_TIMEOUT_MS); + const interrupt = yield* Queue.take(fixture.promptCommands); + assert.deepStrictEqual( + { type: interrupt.type, runId: "runId" in interrupt ? interrupt.runId : null }, + { type: "run.interrupt", runId: RunId.make(`run:${thread.id}`) }, + ); + assert.strictEqual((yield* Queue.take(fixture.promptCommands)).type, "thread.settle"); + }).pipe(Effect.provide(fixture.layer)); + }), + ), + ); + }); }); describe("ThreadSettlementServiceV2 single-thread sweeps", () => { diff --git a/apps/server/src/orchestration-v2/ThreadSettlementService.ts b/apps/server/src/orchestration-v2/ThreadSettlementService.ts index 8b55e25cdc39..491fcaf17c52 100644 --- a/apps/server/src/orchestration-v2/ThreadSettlementService.ts +++ b/apps/server/src/orchestration-v2/ThreadSettlementService.ts @@ -3,7 +3,10 @@ import { resolveProjectSettings } from "@t3tools/shared/projectSettings"; import { visibleThreadPullRequests } from "@t3tools/shared/threadPullRequests"; import { CommandId, + MessageId, + type RunId, type ThreadId, + type OrchestrationV2AppThread, type OrchestrationV2DomainEvent, type OrchestrationV2ThreadShell, } from "@t3tools/contracts"; @@ -37,6 +40,15 @@ export interface SettlementPullRequest { const DAY_MS = 24 * 60 * 60 * 1_000; export const QUEUED_TURN_START_GRACE_MS = 2 * 60 * 1_000; +/** How long a settle prompt's turn may run before it is stopped and the thread settles. */ +export const SETTLE_PROMPT_TIMEOUT_MS = 10 * 60 * 1_000; +const TERMINAL_RUN_STATUSES: ReadonlySet = new Set([ + "completed", + "interrupted", + "failed", + "cancelled", + "rolled_back", +]); function toMillis(value: DateTime.Utc | null | undefined): number | null { return value == null ? null : DateTime.toEpochMillis(value); @@ -556,6 +568,135 @@ export const make = Effect.gen(function* () { runSweep(null, threadId), ); + // Settle prompt: a thread whose project sets one gets that message as one + // more turn when it settles (sending it re-engages the thread), then settles + // again when the turn ends. "running" waits for that turn; "settling" marks + // the settle this service asked for, so it cleans up instead of prompting. + const settlePrompts = new Map< + ThreadId, + { readonly messageId: MessageId; phase: "running" | "settling"; runId: RunId | null } + >(); + // Settling a settled thread re-emits thread.settled with the same settledAt; + // one prompt per settlement. + const settlePromptedAt = new Map(); + + const settleAfterPrompt = Effect.fn("ThreadSettlementServiceV2.settleAfterPrompt")( + function* (threadId: ThreadId, messageId: MessageId, timedOut: boolean) { + const pending = settlePrompts.get(threadId); + if (pending === undefined || pending.messageId !== messageId || pending.phase !== "running") { + return; + } + pending.phase = "settling"; + const uuid = yield* crypto.randomUUIDv4; + // A turn still running at the timeout is stopped; settling refuses active work. + if (timedOut && pending.runId !== null) { + yield* orchestrator.dispatch({ + type: "run.interrupt", + commandId: CommandId.make(`server:settle-prompt-interrupt:${threadId}:${uuid}`), + threadId, + runId: pending.runId, + }); + } + yield* orchestrator + .dispatch({ + type: "thread.settle", + commandId: CommandId.make(`server:settle-prompt-settle:${threadId}:${uuid}`), + threadId, + }) + .pipe( + // Refused when the turn left the thread blocked on the user (a question, + // an approval, a queued message): it stays active for them. + Effect.tapError(() => Effect.sync(() => settlePrompts.delete(threadId))), + ); + }, + (effect, threadId) => + effect.pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("settling after the settle prompt failed", { + threadId, + cause: Cause.pretty(cause), + }), + ), + ), + ); + + /** Sends the project's settle prompt; false when there is none to send. */ + const sendSettlePrompt = Effect.fn("ThreadSettlementServiceV2.sendSettlePrompt")( + function* (thread: OrchestrationV2AppThread) { + if (thread.archivedAt !== null) return false; + const settledAtMs = toMillis(thread.settledAt); + if (settledAtMs === null || settlePromptedAt.get(thread.id) === settledAtMs) return false; + const prompt = resolveProjectSettings(yield* settingsService.getSettings, thread.projectId) + .settings.settlePrompt; + if (prompt === "") return false; + settlePromptedAt.set(thread.id, settledAtMs); + const uuid = yield* crypto.randomUUIDv4; + const messageId = MessageId.make(`message:settle-prompt:${uuid}`); + settlePrompts.set(thread.id, { messageId, phase: "running", runId: null }); + yield* orchestrator + .dispatch({ + type: "message.dispatch", + commandId: CommandId.make(`server:settle-prompt:${thread.id}:${uuid}`), + threadId: thread.id, + messageId, + text: prompt, + attachments: [], + dispatchMode: { type: "start_immediately" }, + createdBy: "system", + creationSource: "server", + }) + .pipe(Effect.tapError(() => Effect.sync(() => settlePrompts.delete(thread.id)))); + yield* Effect.sleep(SETTLE_PROMPT_TIMEOUT_MS).pipe( + Effect.andThen(settleAfterPrompt(thread.id, messageId, true)), + Effect.forkDetach, + ); + return true; + }, + (effect, thread) => + effect.pipe( + Effect.catchCause((cause) => + Cause.hasInterruptsOnly(cause) + ? Effect.failCause(cause) + : Effect.logWarning("sending the settle prompt failed", { + threadId: thread.id, + cause: Cause.pretty(cause), + }).pipe(Effect.as(false)), + ), + ), + ); + + const onThreadSettled = Effect.fn("ThreadSettlementServiceV2.onThreadSettled")(function* ( + thread: OrchestrationV2AppThread, + ) { + const pending = settlePrompts.get(thread.id); + if (pending?.phase === "running") return; + if (pending?.phase === "settling") { + // The settle that follows the prompt: the thread is done, clean up as usual. + settlePrompts.delete(thread.id); + } else if (yield* sendSettlePrompt(thread)) { + // Cleanup waits for the settle that follows the prompt's turn. + return; + } + yield* cleanUpSettledThread(thread.id); + }); + + /** A settle prompt's turn ended (any way): settle the thread again. */ + const onRunUpdated = (run: { + readonly id: RunId; + readonly threadId: ThreadId; + readonly userMessageId: MessageId; + readonly status: string; + }) => { + const pending = settlePrompts.get(run.threadId); + if (pending === undefined || pending.messageId !== run.userMessageId) return Effect.void; + pending.runId = run.id; + return TERMINAL_RUN_STATUSES.has(run.status) + ? settleAfterPrompt(run.threadId, pending.messageId, false) + : Effect.void; + }; + // Settling closes the thread's shells that sit at an idle prompt, so they stop // holding the worktree. A terminal running a command (a dev server, an // editor) stays for the user to close. Then the project's settle script runs @@ -605,7 +746,7 @@ export const make = Effect.gen(function* () { const processEvent = (event: OrchestrationV2DomainEvent) => { switch (event.type) { case "thread.settled": - return cleanUpSettledThread(event.threadId); + return onThreadSettled(event.payload); case "thread.pull-request-synced": case "provider-session.detached": return worker.enqueue(event.threadId); @@ -613,12 +754,16 @@ export const make = Effect.gen(function* () { return event.payload.status !== "starting" && event.payload.status !== "running" ? worker.enqueue(event.threadId) : Effect.void; + case "run.created": + return onRunUpdated(event.payload); case "run.updated": - return ["completed", "interrupted", "failed", "cancelled", "rolled_back"].includes( - event.payload.status, - ) - ? worker.enqueue(event.threadId) - : Effect.void; + return onRunUpdated(event.payload).pipe( + Effect.andThen( + TERMINAL_RUN_STATUSES.has(event.payload.status) + ? worker.enqueue(event.threadId) + : Effect.void, + ), + ); default: return Effect.void; } diff --git a/apps/web/src/components/settings/SettingsPanels.tsx b/apps/web/src/components/settings/SettingsPanels.tsx index bee57d21b253..b064e4a04cba 100644 --- a/apps/web/src/components/settings/SettingsPanels.tsx +++ b/apps/web/src/components/settings/SettingsPanels.tsx @@ -2,6 +2,7 @@ import { SettingsGroup } from "./SettingsGroup"; import { useScopedSettingsWriteAllowed } from "./useScopedSettings"; import { Spinner } from "~/components/ui/spinner"; import { NotificationSettings } from "./NotificationSettings"; +import { SettlePromptSettings } from "./SettlePromptSettings"; import { PRIVACY_POLICY_URL } from "../../legalLinks"; import { ArchiveIcon, ArchiveX, CheckIcon, ChevronRightIcon, SettingsIcon } from "lucide-react"; import { Link, useNavigate } from "@tanstack/react-router"; @@ -584,6 +585,9 @@ export function useSettingsRestore(onRestored?: () => void) { ...(settings.sidebarAutoSettleOnMerge !== DEFAULT_UNIFIED_SETTINGS.sidebarAutoSettleOnMerge ? ["Auto-settle merged threads"] : []), + ...(settings.settlePrompt !== DEFAULT_UNIFIED_SETTINGS.settlePrompt + ? ["Message on settle"] + : []), ...(settings.autoResumeLimitedThreads !== DEFAULT_UNIFIED_SETTINGS.autoResumeLimitedThreads ? ["Auto-resume limited threads"] : []), @@ -708,6 +712,7 @@ export function useSettingsRestore(onRestored?: () => void) { settings.continueThreadsAfterServerUpdate, settings.sidebarAutoSettleAfterDays, settings.sidebarAutoSettleOnMerge, + settings.settlePrompt, settings.autoResumeLimitedThreads, settings.snoozeLimitedThreads, settings.sidebarProjectGroupingMode, @@ -815,6 +820,7 @@ export function useSettingsRestore(onRestored?: () => void) { sidebarWorkingShelfEnabled: DEFAULT_UNIFIED_SETTINGS.sidebarWorkingShelfEnabled, sidebarAutoSettleAfterDays: DEFAULT_UNIFIED_SETTINGS.sidebarAutoSettleAfterDays, sidebarAutoSettleOnMerge: DEFAULT_UNIFIED_SETTINGS.sidebarAutoSettleOnMerge, + settlePrompt: DEFAULT_UNIFIED_SETTINGS.settlePrompt, autoResumeLimitedThreads: DEFAULT_UNIFIED_SETTINGS.autoResumeLimitedThreads, snoozeLimitedThreads: DEFAULT_UNIFIED_SETTINGS.snoozeLimitedThreads, responseStreamingMode: DEFAULT_UNIFIED_SETTINGS.responseStreamingMode, @@ -2477,6 +2483,7 @@ export function GeneralSettingsPanel() { } /> ) : null} + ) : null} diff --git a/apps/web/src/components/settings/SettlePromptSettings.tsx b/apps/web/src/components/settings/SettlePromptSettings.tsx new file mode 100644 index 000000000000..1f79d3c304f9 --- /dev/null +++ b/apps/web/src/components/settings/SettlePromptSettings.tsx @@ -0,0 +1,61 @@ +import { useRef } from "react"; + +import { Textarea } from "../ui/textarea"; +import { SettingsRow, SettingResetButton } from "./settingsLayout"; +import { useSettingsScope } from "./SettingsScopeContext"; +import { searchableSetting } from "./settingsSearch"; +import { + useScopedSettings, + useScopedSettingsMixed, + useUpdateScopedSettings, +} from "./useScopedSettings"; + +/** A message, such as a skill, sent to a thread as one more turn when it settles. */ +export function SettlePromptSettings() { + const settings = useScopedSettings(); + const { targets } = useSettingsScope(); + const scopeKey = targets.map((target) => `${target.environmentId}:${target.projectId}`).join(","); + const edited = useRef(false); + const updateSettings = useUpdateScopedSettings(); + const mixed = useScopedSettingsMixed(["settlePrompt"]); + + return ( + updateSettings({ settlePrompt: "" })} + /> + ) : null + } + > +
+