diff --git a/apps/mobile/src/features/home/HomeScreen.tsx b/apps/mobile/src/features/home/HomeScreen.tsx index b738ff6dd12f..a199bfb5568b 100644 --- a/apps/mobile/src/features/home/HomeScreen.tsx +++ b/apps/mobile/src/features/home/HomeScreen.tsx @@ -628,6 +628,15 @@ export function HomeScreen(props: HomeScreenProps) { } return supported; }, [serverConfigs]); + const persistenceEnvironmentIds = useMemo(() => { + const supported = new Set(); + for (const [environmentId, config] of serverConfigs) { + if (config.environment.capabilities.threadPersistence === true) { + supported.add(environmentId); + } + } + return supported; + }, [serverConfigs]); const pinReorderEnvironmentIds = useMemo(() => { const supported = new Set(); for (const [environmentId, config] of serverConfigs) { @@ -848,6 +857,7 @@ export function HomeScreen(props: HomeScreenProps) { onArchiveThread={props.onArchiveThread} onRegenerateThreadTitle={handleRegenerateThreadTitle} titleRegenerationSupported={titleRegenerationEnvironmentIds.has(thread.environmentId)} + persistenceSupported={persistenceEnvironmentIds.has(thread.environmentId)} settlementSupported={settlementEnvironmentIds.has(thread.environmentId)} onSettleThread={handleSettleThread} snoozeSupported={snoozeEnvironmentIds.has(thread.environmentId)} @@ -888,6 +898,7 @@ export function HomeScreen(props: HomeScreenProps) { handleSwipeableWillOpen, handleUnsettleThread, pinningEnvironmentIds, + persistenceEnvironmentIds, pinReorderEnvironmentIds, projectByKey, projectCwdByKey, @@ -1009,6 +1020,7 @@ export function HomeScreen(props: HomeScreenProps) { onDeleteThread={props.onDeleteThread} onRegenerateThreadTitle={handleRegenerateThreadTitle} titleRegenerationSupported={titleRegenerationEnvironmentIds.has(thread.environmentId)} + persistenceSupported={persistenceEnvironmentIds.has(thread.environmentId)} onSelectThread={props.onSelectThread} onSwipeableClose={handleSwipeableClose} onSwipeableWillOpen={handleSwipeableWillOpen} @@ -1040,6 +1052,7 @@ export function HomeScreen(props: HomeScreenProps) { props.onSelectThread, props.searchQuery, props.savedConnectionsById, + persistenceEnvironmentIds, threadSearchMatchByKey, titleRegenerationEnvironmentIds, updateGroupDisplay, diff --git a/apps/mobile/src/features/threads/PersistentThreadIcon.tsx b/apps/mobile/src/features/threads/PersistentThreadIcon.tsx new file mode 100644 index 000000000000..3511fe02e6af --- /dev/null +++ b/apps/mobile/src/features/threads/PersistentThreadIcon.tsx @@ -0,0 +1,23 @@ +import Svg, { Path, Rect } from "react-native-svg"; + +export function PersistentThreadIcon(props: { readonly color: string; readonly size?: number }) { + const size = props.size ?? 14; + return ( + + + + + + ); +} diff --git a/apps/mobile/src/features/threads/ThreadNavigationSidebar.tsx b/apps/mobile/src/features/threads/ThreadNavigationSidebar.tsx index 512b9b78a4ca..5800e906485d 100644 --- a/apps/mobile/src/features/threads/ThreadNavigationSidebar.tsx +++ b/apps/mobile/src/features/threads/ThreadNavigationSidebar.tsx @@ -461,6 +461,15 @@ function ThreadNavigationSidebarPane( } return supported; }, [serverConfigs]); + const persistenceEnvironmentIds = useMemo(() => { + const supported = new Set(); + for (const [environmentId, config] of serverConfigs) { + if (config.environment.capabilities.threadPersistence === true) { + supported.add(environmentId); + } + } + return supported; + }, [serverConfigs]); const pinReorderEnvironmentIds = useMemo(() => { const supported = new Set(); for (const [environmentId, config] of serverConfigs) { @@ -913,6 +922,7 @@ function ThreadNavigationSidebarPane( onArchiveThread={archiveThread} onRegenerateThreadTitle={regenerateThreadTitle} titleRegenerationSupported={titleRegenerationEnvironmentIds.has(thread.environmentId)} + persistenceSupported={persistenceEnvironmentIds.has(thread.environmentId)} settlementSupported={settlementEnvironmentIds.has(thread.environmentId)} onSettleThread={settleThread} snoozeSupported={snoozeEnvironmentIds.has(thread.environmentId)} @@ -1034,6 +1044,7 @@ function ThreadNavigationSidebarPane( onDeleteThread={confirmDeleteThread} onRegenerateThreadTitle={regenerateThreadTitle} titleRegenerationSupported={titleRegenerationEnvironmentIds.has(thread.environmentId)} + persistenceSupported={persistenceEnvironmentIds.has(thread.environmentId)} onSelectThread={handleSelectThread} onSwipeableClose={handleSwipeableClose} onSwipeableWillOpen={handleSwipeableWillOpen} @@ -1067,6 +1078,7 @@ function ThreadNavigationSidebarPane( pinReorderEnvironmentIds, pinThread, pinningEnvironmentIds, + persistenceEnvironmentIds, projectByKey, projectCwdByKey, projectTitleByProjectKey, diff --git a/apps/mobile/src/features/threads/thread-list-items.tsx b/apps/mobile/src/features/threads/thread-list-items.tsx index 024e02d2ea0c..e704e32a675b 100644 --- a/apps/mobile/src/features/threads/thread-list-items.tsx +++ b/apps/mobile/src/features/threads/thread-list-items.tsx @@ -32,6 +32,11 @@ import { buildThreadTitleRegenerationMenuItems } from "./thread-title-regenerati import { resolveThreadStatus, shouldShowActionWaitingIndicator } from "./threadPresentation"; import { actionRunningPresentation } from "@t3tools/shared/actionResume"; import { ThreadSearchMatchExcerpt } from "./thread-search-match"; +import { PersistentThreadIcon } from "./PersistentThreadIcon"; +import { + buildThreadPersistenceMenuItems, + persistenceIntentForMenuEvent, +} from "./thread-persistence-menu"; /** * Shared presentation for the thread lists: the compact (phone) Home list and @@ -439,6 +444,7 @@ export const ThreadListRow = memo(function ThreadListRow(props: { readonly onDeleteThread: (thread: EnvironmentThreadShell) => void; readonly onRegenerateThreadTitle: (thread: EnvironmentThreadShell) => void; readonly titleRegenerationSupported: boolean; + readonly persistenceSupported: boolean; readonly onSwipeableWillOpen: (methods: SwipeableMethods) => void; readonly onSwipeableClose: (methods: SwipeableMethods) => void; readonly simultaneousSwipeGesture?: ComponentProps< @@ -473,6 +479,9 @@ export const ThreadListRow = memo(function ThreadListRow(props: { const abandonWorktreeCleanup = useAtomCommand(threadEnvironment.abandonWorktreeCleanup, { reportFailure: false, }); + const setThreadPersistence = useAtomCommand(threadEnvironment.setPersistence, { + reportFailure: false, + }); const status = resolveThreadStatus(thread); const pr = useThreadPr(thread, props.projectCwd); const timestamp = relativeTime( @@ -511,6 +520,22 @@ export const ThreadListRow = memo(function ThreadListRow(props: { () => onRegenerateThreadTitle(thread), [onRegenerateThreadTitle, thread], ); + const handlePersistence = useCallback( + async (persistent: boolean) => { + const result = await setThreadPersistence({ + environmentId: thread.environmentId, + input: { threadId: thread.id, persistent }, + }); + if (result._tag === "Failure") { + const error = Cause.squash(result.cause); + Alert.alert( + "Could not update persistent thread", + error instanceof Error ? error.message : "The persistent thread could not be updated.", + ); + } + }, + [setThreadPersistence, thread.environmentId, thread.id], + ); const handleCancelAction = useCallback(async () => { if (runningAction === null) return; const result = await closeTerminal({ @@ -568,25 +593,36 @@ export const ThreadListRow = memo(function ThreadListRow(props: { ); }, [abandonWorktreeCleanup, thread.environmentId, thread.id]); const menuActions = useMemo( - () => [ - THREAD_ROW_MENU_ACTIONS[0]!, - ...buildThreadTitleRegenerationMenuItems({ - supported: props.titleRegenerationSupported, - isRegenerating: thread.titleRegeneration != null, + () => + buildThreadPersistenceMenuItems({ + persistent: thread.persistent === true, + supported: props.persistenceSupported, + actions: [ + THREAD_ROW_MENU_ACTIONS[0]!, + ...buildThreadTitleRegenerationMenuItems({ + supported: props.titleRegenerationSupported, + isRegenerating: thread.titleRegeneration != null, + }), + ...(runningAction === null + ? [] + : [ + { + id: "cancel-action", + title: `Cancel ${runningAction.actionName}`, + image: "stop.fill", + attributes: { destructive: true }, + } satisfies MenuAction, + ]), + THREAD_ROW_MENU_ACTIONS[1]!, + ], }), - ...(runningAction === null - ? [] - : [ - { - id: "cancel-action", - title: `Cancel ${runningAction.actionName}`, - image: "stop.fill", - attributes: { destructive: true }, - } satisfies MenuAction, - ]), - THREAD_ROW_MENU_ACTIONS[1]!, + [ + props.persistenceSupported, + props.titleRegenerationSupported, + runningAction, + thread.persistent, + thread.titleRegeneration, ], - [props.titleRegenerationSupported, runningAction, thread.titleRegeneration], ); const primaryAction = useMemo( () => ({ @@ -605,12 +641,15 @@ export const ThreadListRow = memo(function ThreadListRow(props: { if (nativeEvent.event === "retry-worktree-cleanup") void handleRetryWorktreeCleanup(); if (nativeEvent.event === "keep-worktree") handleKeepWorktree(); if (nativeEvent.event === "delete") handleDelete(); + const persistenceIntent = persistenceIntentForMenuEvent(nativeEvent.event); + if (persistenceIntent !== null) void handlePersistence(persistenceIntent); }, [ handleArchive, handleCancelAction, handleDelete, handleKeepWorktree, + handlePersistence, handleRegenerateTitle, handleRetryWorktreeCleanup, ], @@ -679,7 +718,9 @@ export const ThreadListRow = memo(function ThreadListRow(props: { accessibilityHint={ cleanupFailed ? "Thread unavailable. Long-press for worktree recovery actions" - : "Swipe left for archive and delete actions" + : thread.persistent === true + ? "Persistent thread. Long-press to disable persistence" + : "Swipe left for archive and delete actions" } accessibilityLabel={threadAccessibilityLabel} accessibilityRole="button" @@ -709,9 +750,20 @@ export const ThreadListRow = memo(function ThreadListRow(props: { }} > - - {thread.title} - + + {thread.persistent === true ? ( + + ) : null} + + {thread.title} + + {actionStatusIndicator} {statusPill} @@ -769,15 +821,23 @@ export const ThreadListRow = memo(function ThreadListRow(props: { > - - {thread.title} - + + {thread.persistent === true ? ( + + ) : null} + + {thread.title} + + {actionStatusIndicator} {statusPill} @@ -807,7 +867,7 @@ export const ThreadListRow = memo(function ThreadListRow(props: { return ( { + const result = await setThreadPersistence({ + environmentId: thread.environmentId, + input: { threadId: thread.id, persistent }, + }); + if (result._tag === "Failure") { + const error = Cause.squash(result.cause); + Alert.alert( + "Could not update persistent thread", + error instanceof Error ? error.message : "The persistent thread could not be updated.", + ); + } + }, + [setThreadPersistence, thread.environmentId, thread.id], + ); // Swipe: the v2 primary action is the lifecycle transition. Every settled // row can un-settle — explicit settles clear the override, auto-settled @@ -641,59 +667,79 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { ], [runningAction], ); + const withPersistence = useCallback( + (actions: ReadonlyArray) => + buildThreadPersistenceMenuItems({ + actions, + persistent: thread.persistent === true, + supported: props.persistenceSupported, + }), + [props.persistenceSupported, thread.persistent], + ); const snoozableCardMenuActions = useMemo( - () => [ - { id: "settle", title: "Settle", image: "checkmark" }, - { - id: "snooze", - title: "Snooze", - image: "clock", - subactions: snoozePresetActions, - }, - ...pinMenuItem, - ...titleRegenerationMenuItems, - ...actionMenuItems, - { id: "delete", title: "Delete", image: "trash", attributes: { destructive: true } }, + () => + withPersistence([ + { id: "settle", title: "Settle", image: "checkmark" }, + { + id: "snooze", + title: "Snooze", + image: "clock", + subactions: snoozePresetActions, + }, + ...pinMenuItem, + ...titleRegenerationMenuItems, + ...actionMenuItems, + { id: "delete", title: "Delete", image: "trash", attributes: { destructive: true } }, + ]), + [ + actionMenuItems, + pinMenuItem, + snoozePresetActions, + titleRegenerationMenuItems, + withPersistence, ], - [actionMenuItems, pinMenuItem, snoozePresetActions, titleRegenerationMenuItems], ); const cardMenuActions = useMemo( - () => [ - CARD_MENU_ACTIONS[0]!, - ...pinMenuItem, - ...titleRegenerationMenuItems, - ...actionMenuItems, - ...CARD_MENU_ACTIONS.slice(1), - ], - [actionMenuItems, pinMenuItem, titleRegenerationMenuItems], + () => + withPersistence([ + CARD_MENU_ACTIONS[0]!, + ...pinMenuItem, + ...titleRegenerationMenuItems, + ...actionMenuItems, + ...CARD_MENU_ACTIONS.slice(1), + ]), + [actionMenuItems, pinMenuItem, titleRegenerationMenuItems, withPersistence], ); const slimMenuActions = useMemo( - () => [ - SLIM_MENU_ACTIONS[0]!, - ...(thread.pinnedAt != null ? pinMenuItem : []), - ...titleRegenerationMenuItems, - ...actionMenuItems, - SLIM_MENU_ACTIONS[1]!, - ], - [actionMenuItems, pinMenuItem, thread.pinnedAt, titleRegenerationMenuItems], + () => + withPersistence([ + SLIM_MENU_ACTIONS[0]!, + ...(thread.pinnedAt != null ? pinMenuItem : []), + ...titleRegenerationMenuItems, + ...actionMenuItems, + SLIM_MENU_ACTIONS[1]!, + ]), + [actionMenuItems, pinMenuItem, thread.pinnedAt, titleRegenerationMenuItems, withPersistence], ); const snoozedMenuActions = useMemo( - () => [ - SNOOZED_MENU_ACTIONS[0]!, - ...titleRegenerationMenuItems, - ...actionMenuItems, - SNOOZED_MENU_ACTIONS[1]!, - ], - [actionMenuItems, titleRegenerationMenuItems], + () => + withPersistence([ + SNOOZED_MENU_ACTIONS[0]!, + ...titleRegenerationMenuItems, + ...actionMenuItems, + SNOOZED_MENU_ACTIONS[1]!, + ]), + [actionMenuItems, titleRegenerationMenuItems, withPersistence], ); const legacyMenuActions = useMemo( - () => [ - LEGACY_MENU_ACTIONS[0]!, - ...titleRegenerationMenuItems, - ...actionMenuItems, - LEGACY_MENU_ACTIONS[1]!, - ], - [actionMenuItems, titleRegenerationMenuItems], + () => + withPersistence([ + LEGACY_MENU_ACTIONS[0]!, + ...titleRegenerationMenuItems, + ...actionMenuItems, + LEGACY_MENU_ACTIONS[1]!, + ]), + [actionMenuItems, titleRegenerationMenuItems, withPersistence], ); const handleMenuAction = useCallback( ({ nativeEvent }: { readonly nativeEvent: { readonly event: string } }) => { @@ -710,6 +756,8 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { if (nativeEvent.event === "retry-worktree-cleanup") void handleRetryWorktreeCleanup(); if (nativeEvent.event === "keep-worktree") handleKeepWorktree(); if (nativeEvent.event === "delete") handleDelete(); + const persistenceIntent = persistenceIntentForMenuEvent(nativeEvent.event); + if (persistenceIntent !== null) void handlePersistence(persistenceIntent); const snoozeSelection = resolveThreadListV2SnoozeMenuSelection({ event: nativeEvent.event, displayedPresets: snoozePresets, @@ -726,6 +774,7 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { handleCancelAction, handleDelete, handleKeepWorktree, + handlePersistence, handleRegenerateTitle, handleRetryWorktreeCleanup, handleMovePinnedDown, @@ -855,15 +904,25 @@ export const ThreadListV2Row = memo(function ThreadListV2Row(props: { - - {thread.title} - + + {thread.persistent === true ? ( + + + + ) : null} + + {thread.title} + + {props.searchMatch ? ( ) : null} - - {thread.title} - + + {thread.persistent === true ? ( + + ) : null} + + {thread.title} + + {props.searchMatch ? ( { + const actions = [ + { id: "archive", title: "Archive" }, + { id: "delete", title: "Delete", attributes: { destructive: true } }, + ] as const; + + it("offers designation when supported", () => { + expect( + buildThreadPersistenceMenuItems({ actions, persistent: false, supported: true })[0], + ).toMatchObject({ id: "mark-persistent", title: "Mark as persistent thread" }); + }); + + it("offers disable and guards destructive lifecycle actions", () => { + const items = buildThreadPersistenceMenuItems({ actions, persistent: true, supported: true }); + + expect(items[0]).toMatchObject({ id: "disable-persistence" }); + expect(items.slice(1)).toEqual([ + { + id: "archive", + title: "Archive (disable persistence first)", + attributes: { disabled: true }, + }, + { + id: "delete", + title: "Delete (disable persistence first)", + attributes: { destructive: true, disabled: true }, + }, + ]); + }); + + it("preserves the selected persistence intent even if shell state changes", () => { + expect(persistenceIntentForMenuEvent("mark-persistent")).toBe(true); + expect(persistenceIntentForMenuEvent("disable-persistence")).toBe(false); + expect(persistenceIntentForMenuEvent("archive")).toBeNull(); + }); +}); diff --git a/apps/mobile/src/features/threads/thread-persistence-menu.ts b/apps/mobile/src/features/threads/thread-persistence-menu.ts new file mode 100644 index 000000000000..14440616248e --- /dev/null +++ b/apps/mobile/src/features/threads/thread-persistence-menu.ts @@ -0,0 +1,32 @@ +import type { MenuAction } from "@react-native-menu/menu"; + +export function persistenceIntentForMenuEvent(event: string): boolean | null { + if (event === "mark-persistent") return true; + if (event === "disable-persistence") return false; + return null; +} + +export function buildThreadPersistenceMenuItems(input: { + readonly actions: ReadonlyArray; + readonly persistent: boolean; + readonly supported: boolean; +}): MenuAction[] { + const protectedActions = input.actions.map((action) => + input.persistent && (action.id === "archive" || action.id === "delete") + ? { + ...action, + title: `${action.title} (disable persistence first)`, + attributes: { ...action.attributes, disabled: true }, + } + : action, + ); + if (!input.supported) return protectedActions; + return [ + { + id: input.persistent ? "disable-persistence" : "mark-persistent", + title: input.persistent ? "Disable persistent thread" : "Mark as persistent thread", + image: "lock", + }, + ...protectedActions, + ]; +} diff --git a/apps/server/src/cli/thread.ts b/apps/server/src/cli/thread.ts index e3a413537d57..ff27441dc089 100644 --- a/apps/server/src/cli/thread.ts +++ b/apps/server/src/cli/thread.ts @@ -107,6 +107,7 @@ export const ThreadListResult = Schema.Struct({ Schema.Struct({ ...ThreadIdentity.fields, title: Schema.String, + persistent: Schema.Boolean, lifecycle: Schema.String, project: ThreadProject, workspace: ThreadWorkspace, @@ -120,6 +121,7 @@ export const ThreadReadResult = Schema.Struct({ kind: Schema.Literal("read"), ...ThreadIdentity.fields, title: Schema.String, + persistent: Schema.Boolean, lifecycle: Schema.String, project: ThreadProject, workspace: ThreadWorkspace, @@ -621,7 +623,9 @@ export const listThreadsOutput = Effect.fn("listThreadsOutput")(function* ( const now = DateTime.formatIso(yield* DateTime.now); const sortedThreads = source.shell.threads.toSorted( (left, right) => - right.updatedAt.localeCompare(left.updatedAt) || left.id.localeCompare(right.id), + Number(right.persistent === true) - Number(left.persistent === true) || + right.updatedAt.localeCompare(left.updatedAt) || + left.id.localeCompare(right.id), ); const threadsTruncated = sortedThreads.length > THREAD_LIST_MAX_RESULTS; return yield* decodeThreadListResult({ @@ -636,6 +640,7 @@ export const listThreadsOutput = Effect.fn("listThreadsOutput")(function* ( environmentId: source.descriptor.environmentId, threadId: thread.id, title: thread.title, + persistent: thread.persistent === true, lifecycle: threadLifecycle(thread, { now }), project: projectOutput(project), workspace: workspaceOutput(project, thread), @@ -665,6 +670,7 @@ export const readThreadOutput = Effect.fn("readThreadOutput")(function* ( environmentId: source.descriptor.environmentId, threadId: resolution.thread.id, title: resolution.thread.title, + persistent: resolution.thread.persistent === true, lifecycle: threadLifecycle(resolution.thread, { now }), project: projectOutput(project), workspace: workspaceOutput(project, resolution.thread), diff --git a/apps/server/src/environment/ServerEnvironment.ts b/apps/server/src/environment/ServerEnvironment.ts index fe5b61d50468..bb00bccb4b5f 100644 --- a/apps/server/src/environment/ServerEnvironment.ts +++ b/apps/server/src/environment/ServerEnvironment.ts @@ -157,6 +157,7 @@ export const make = Effect.gen(function* () { threadSnooze: true, environmentThemes: true, threadPinning: true, + threadPersistence: true, threadPinReorder: true, threadTitleRegeneration: true, threadPullRequestLinking: true, diff --git a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts index e6b05e968485..af1297251a25 100644 --- a/apps/server/src/orchestration/Layers/ProjectionPipeline.ts +++ b/apps/server/src/orchestration/Layers/ProjectionPipeline.ts @@ -656,6 +656,7 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti snoozedUntil: null, snoozedAt: null, pinnedAt: null, + persistent: 0, pinOrderKey: null, titleRegenerationRequestId: null, titleRegenerationStartedAt: null, @@ -702,6 +703,22 @@ const makeOrchestrationProjectionPipeline = Effect.fn("makeOrchestrationProjecti return; } + case "thread.persistence-changed": { + const affectedIds = [event.payload.threadId, event.payload.replacedThreadId].filter( + (threadId): threadId is ThreadId => threadId !== null, + ); + for (const threadId of new Set(affectedIds)) { + const existingRow = yield* projectionThreadRepository.getById({ threadId }); + if (Option.isNone(existingRow)) continue; + yield* projectionThreadRepository.upsert({ + ...existingRow.value, + persistent: threadId === event.payload.persistentThreadId ? 1 : 0, + updatedAt: event.payload.updatedAt, + }); + } + return; + } + case "thread.settled": { const existingRow = yield* projectionThreadRepository.getById({ threadId: event.payload.threadId, diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts index 2556d88f82ac..261a2b3bbb20 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.test.ts @@ -341,6 +341,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { snoozedUntil: null, snoozedAt: null, pinnedAt: "2026-02-24T00:00:01.000Z", + persistent: false, pinOrderKey: "gm", titleRegeneration: null, annotation: { @@ -475,6 +476,7 @@ projectionSnapshotLayer("ProjectionSnapshotQuery", (it) => { snoozedUntil: null, snoozedAt: null, pinnedAt: "2026-02-24T00:00:01.000Z", + persistent: false, pinOrderKey: "gm", titleRegeneration: null, annotation: { diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts index 5c5a2905e63d..f79545ad60d4 100644 --- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts +++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts @@ -447,6 +447,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", pinned_at AS "pinnedAt", + persistent, pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", @@ -488,6 +489,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", pinned_at AS "pinnedAt", + persistent, pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", @@ -531,6 +533,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", pinned_at AS "pinnedAt", + persistent, pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", @@ -979,6 +982,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", pinned_at AS "pinnedAt", + persistent, pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", @@ -1749,6 +1753,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, pinnedAt: row.pinnedAt, + persistent: (row.persistent ?? 0) > 0, pinOrderKey: row.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(row), ...(row.annotation !== null ? { annotation: row.annotation } : {}), @@ -1962,6 +1967,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, pinnedAt: row.pinnedAt, + persistent: (row.persistent ?? 0) > 0, pinOrderKey: row.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(row), ...(row.annotation !== null ? { annotation: row.annotation } : {}), @@ -2105,6 +2111,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, pinnedAt: row.pinnedAt, + persistent: (row.persistent ?? 0) > 0, pinOrderKey: row.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(row), ...(row.annotation !== null ? { annotation: row.annotation } : {}), @@ -2259,6 +2266,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { snoozedUntil: row.snoozedUntil, snoozedAt: row.snoozedAt, pinnedAt: row.pinnedAt, + persistent: (row.persistent ?? 0) > 0, pinOrderKey: row.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(row), ...(row.annotation !== null ? { annotation: row.annotation } : {}), @@ -2545,6 +2553,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { snoozedUntil: threadRow.value.snoozedUntil, snoozedAt: threadRow.value.snoozedAt, pinnedAt: threadRow.value.pinnedAt, + persistent: (threadRow.value.persistent ?? 0) > 0, pinOrderKey: threadRow.value.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(threadRow.value), ...(threadRow.value.annotation !== null ? { annotation: threadRow.value.annotation } : {}), @@ -2695,6 +2704,7 @@ const makeProjectionSnapshotQuery = Effect.gen(function* () { snoozedUntil: threadRow.value.snoozedUntil, snoozedAt: threadRow.value.snoozedAt, pinnedAt: threadRow.value.pinnedAt, + persistent: (threadRow.value.persistent ?? 0) > 0, pinOrderKey: threadRow.value.pinOrderKey ?? null, titleRegeneration: mapTitleRegeneration(threadRow.value), ...(threadRow.value.annotation !== null ? { annotation: threadRow.value.annotation } : {}), diff --git a/apps/server/src/orchestration/Schemas.ts b/apps/server/src/orchestration/Schemas.ts index 99123c70b96a..29eb2040182b 100644 --- a/apps/server/src/orchestration/Schemas.ts +++ b/apps/server/src/orchestration/Schemas.ts @@ -11,6 +11,7 @@ import { ThreadDeletedPayload as ContractsThreadDeletedPayloadSchema, ThreadWorktreeCleanupUpdatedPayload as ContractsThreadWorktreeCleanupUpdatedPayloadSchema, ThreadUnarchivedPayload as ContractsThreadUnarchivedPayloadSchema, + ThreadPersistenceChangedPayload as ContractsThreadPersistenceChangedPayloadSchema, ThreadUnsettledPayload as ContractsThreadUnsettledPayloadSchema, ThreadSnoozedPayload as ContractsThreadSnoozedPayloadSchema, ThreadUnsnoozedPayload as ContractsThreadUnsnoozedPayloadSchema, @@ -46,6 +47,7 @@ export const ThreadDeletedPayload = ContractsThreadDeletedPayloadSchema; export const ThreadWorktreeCleanupUpdatedPayload = ContractsThreadWorktreeCleanupUpdatedPayloadSchema; export const ThreadUnarchivedPayload = ContractsThreadUnarchivedPayloadSchema; +export const ThreadPersistenceChangedPayload = ContractsThreadPersistenceChangedPayloadSchema; export const ThreadUnsettledPayload = ContractsThreadUnsettledPayloadSchema; export const ThreadSnoozedPayload = ContractsThreadSnoozedPayloadSchema; export const ThreadUnsnoozedPayload = ContractsThreadUnsnoozedPayloadSchema; diff --git a/apps/server/src/orchestration/decider.delete.test.ts b/apps/server/src/orchestration/decider.delete.test.ts index e7b1f87b7248..8a99f766cbf3 100644 --- a/apps/server/src/orchestration/decider.delete.test.ts +++ b/apps/server/src/orchestration/decider.delete.test.ts @@ -490,6 +490,33 @@ it.layer(NodeServices.layer)("decider deletion flows", (it) => { }), ); + it.effect("rejects force-deleting a project containing the persistent thread", () => + Effect.gen(function* () { + const seeded = yield* seedReadModel; + const readModel = { + ...seeded, + threads: seeded.threads.map((thread) => + thread.id === asThreadId("thread-delete-2") ? { ...thread, persistent: true } : thread, + ), + }; + + const error = yield* Effect.flip( + decideOrchestrationCommand({ + command: { + type: "project.delete", + commandId: asCommandId("cmd-project-delete-persistent"), + projectId: asProjectId("project-delete"), + force: true, + }, + readModel, + }), + ); + + expect(error.message).toContain("thread-delete-2"); + expect(error.message).toContain("Disable persistence or move it to another thread first"); + }), + ); + it.effect("rejects project deletion while a deleted thread is cleaning up its worktree", () => Effect.gen(function* () { const seeded = yield* seedReadModel; diff --git a/apps/server/src/orchestration/decider.persistent.test.ts b/apps/server/src/orchestration/decider.persistent.test.ts new file mode 100644 index 000000000000..094d6639dc98 --- /dev/null +++ b/apps/server/src/orchestration/decider.persistent.test.ts @@ -0,0 +1,87 @@ +import { + CommandId, + ProjectId, + ProviderInstanceId, + ThreadId, + type OrchestrationReadModel, +} from "@t3tools/contracts"; +import * as NodeServices from "@effect/platform-node/NodeServices"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { decideOrchestrationCommand } from "./decider.ts"; + +const NOW = "2026-08-30T00:00:00.000Z"; + +function thread(id: string, persistent = false): OrchestrationReadModel["threads"][number] { + return { + id: ThreadId.make(id), + projectId: ProjectId.make("project-1"), + title: id, + modelSelection: { instanceId: ProviderInstanceId.make("codex"), model: "gpt-5.6" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + latestTurn: null, + createdAt: NOW, + updatedAt: NOW, + archivedAt: null, + settledOverride: null, + settledAt: null, + persistent, + deletedAt: null, + messages: [], + proposedPlans: [], + activities: [], + checkpoints: [], + session: null, + }; +} + +function readModel(threads: OrchestrationReadModel["threads"]): OrchestrationReadModel { + return { snapshotSequence: 0, projects: [], threads, updatedAt: NOW }; +} + +it.layer(NodeServices.layer)("persistent thread decider", (it) => { + it.effect("atomically replaces the persistent thread", () => + Effect.gen(function* () { + const event = yield* decideOrchestrationCommand({ + command: { + type: "thread.persistence.set", + commandId: CommandId.make("cmd-persist"), + threadId: ThreadId.make("thread-new"), + persistent: true, + }, + readModel: readModel([thread("thread-old", true), thread("thread-new")]), + }); + const events = Array.isArray(event) ? event : [event]; + expect(events[0]?.type).toBe("thread.persistence-changed"); + if (events[0]?.type === "thread.persistence-changed") { + expect(events[0].payload).toMatchObject({ + persistentThreadId: "thread-new", + replacedThreadId: "thread-old", + }); + } + }), + ); + + for (const type of ["thread.archive", "thread.delete"] as const) { + it.effect(`blocks ${type} for the persistent thread`, () => + Effect.gen(function* () { + const result = yield* Effect.flip( + decideOrchestrationCommand({ + command: { + type, + commandId: CommandId.make(`cmd-${type}`), + threadId: ThreadId.make("thread-persistent"), + }, + readModel: readModel([thread("thread-persistent", true)]), + }), + ); + expect(result.message).toContain("cannot be"); + expect(result.message).toContain("persistence is disabled or moved"); + }), + ); + } +}); diff --git a/apps/server/src/orchestration/decider.ts b/apps/server/src/orchestration/decider.ts index 5dfec84908fc..f0820f89fc37 100644 --- a/apps/server/src/orchestration/decider.ts +++ b/apps/server/src/orchestration/decider.ts @@ -430,6 +430,13 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }); } const activeThreads = projectThreads.filter((thread) => thread.deletedAt === null); + const persistentThread = activeThreads.find((thread) => thread.persistent); + if (persistentThread !== undefined) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Project '${command.projectId}' cannot be deleted while thread '${persistentThread.id}' is persistent. Disable persistence or move it to another thread first.`, + }); + } if (activeThreads.length > 0 && command.force !== true) { return yield* new OrchestrationCommandInvariantError({ commandType: command.type, @@ -524,6 +531,12 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" command, threadId: command.threadId, }); + if (thread.persistent) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Persistent thread '${thread.id}' cannot be deleted until persistence is disabled or moved to another thread.`, + }); + } const occurredAt = yield* nowIso; // Deletion commands can be retried after the first deleted event has @@ -743,11 +756,17 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" } case "thread.archive": { - yield* requireThreadNotArchived({ + const thread = yield* requireThreadNotArchived({ readModel, command, threadId: command.threadId, }); + if (thread.persistent) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Persistent thread '${thread.id}' cannot be archived until persistence is disabled or moved to another thread.`, + }); + } const occurredAt = yield* nowIso; return { ...(yield* withEventBase({ @@ -787,6 +806,41 @@ export const decideOrchestrationCommand = Effect.fn("decideOrchestrationCommand" }; } + case "thread.persistence.set": { + const thread = yield* requireThread({ readModel, command, threadId: command.threadId }); + if (thread.archivedAt !== null || thread.deletedAt !== null) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Only an active thread can be marked persistent.`, + }); + } + const current = readModel.threads.find( + (candidate) => candidate.deletedAt === null && candidate.persistent, + ); + if (!command.persistent && current?.id !== thread.id) { + return yield* new OrchestrationCommandInvariantError({ + commandType: command.type, + detail: `Thread '${thread.id}' is not the persistent thread.`, + }); + } + const occurredAt = yield* nowIso; + return { + ...(yield* withEventBase({ + aggregateKind: "thread", + aggregateId: command.threadId, + occurredAt, + commandId: command.commandId, + })), + type: "thread.persistence-changed", + payload: { + threadId: command.threadId, + persistentThreadId: command.persistent ? command.threadId : null, + replacedThreadId: current?.id ?? null, + updatedAt: occurredAt, + }, + }; + } + case "thread.settle": { const thread = yield* requireThreadNotArchived({ readModel, diff --git a/apps/server/src/orchestration/projector.persistent.test.ts b/apps/server/src/orchestration/projector.persistent.test.ts new file mode 100644 index 000000000000..0d2aca21edfc --- /dev/null +++ b/apps/server/src/orchestration/projector.persistent.test.ts @@ -0,0 +1,70 @@ +import { + CommandId, + EventId, + ProjectId, + ThreadId, + type OrchestrationEvent, +} from "@t3tools/contracts"; +import { expect, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; + +import { createEmptyReadModel, projectEvent } from "./projector.ts"; + +const NOW = "2026-08-30T00:00:00.000Z"; + +function event(sequence: number, type: OrchestrationEvent["type"], payload: unknown) { + return { + sequence, + eventId: EventId.make(`event-${sequence}`), + type, + aggregateKind: "thread", + aggregateId: ThreadId.make("thread-new"), + occurredAt: NOW, + commandId: CommandId.make(`command-${sequence}`), + causationEventId: null, + correlationId: null, + metadata: {}, + payload, + } as OrchestrationEvent; +} + +const created = (sequence: number, threadId: string) => + event(sequence, "thread.created", { + threadId: ThreadId.make(threadId), + projectId: ProjectId.make("project-1"), + title: threadId, + modelSelection: { provider: "codex", model: "gpt-5.6" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + createdAt: NOW, + updatedAt: NOW, + }); + +it.effect("projects persistent thread replacement atomically", () => + Effect.gen(function* () { + const withOld = yield* projectEvent(createEmptyReadModel(NOW), created(1, "thread-old")); + const withBoth = yield* projectEvent(withOld, created(2, "thread-new")); + const oldPersistent = yield* projectEvent( + withBoth, + event(3, "thread.persistence-changed", { + threadId: ThreadId.make("thread-old"), + persistentThreadId: ThreadId.make("thread-old"), + replacedThreadId: null, + updatedAt: NOW, + }), + ); + const replaced = yield* projectEvent( + oldPersistent, + event(4, "thread.persistence-changed", { + threadId: ThreadId.make("thread-new"), + persistentThreadId: ThreadId.make("thread-new"), + replacedThreadId: ThreadId.make("thread-old"), + updatedAt: NOW, + }), + ); + expect(replaced.threads.find((thread) => thread.id === "thread-old")?.persistent).toBe(false); + expect(replaced.threads.find((thread) => thread.id === "thread-new")?.persistent).toBe(true); + }), +); diff --git a/apps/server/src/orchestration/projector.test.ts b/apps/server/src/orchestration/projector.test.ts index 3a0fb895ea76..4049c070d294 100644 --- a/apps/server/src/orchestration/projector.test.ts +++ b/apps/server/src/orchestration/projector.test.ts @@ -96,6 +96,7 @@ describe("orchestration projector", () => { unsettledAt: null, snoozedUntil: null, snoozedAt: null, + persistent: false, deletedAt: null, messages: [], proposedPlans: [], diff --git a/apps/server/src/orchestration/projector.ts b/apps/server/src/orchestration/projector.ts index f4a79b0f731b..0cab426972bf 100644 --- a/apps/server/src/orchestration/projector.ts +++ b/apps/server/src/orchestration/projector.ts @@ -21,6 +21,7 @@ import { ProjectMetaUpdatedPayload, ThreadActivityAppendedPayload, ThreadArchivedPayload, + ThreadPersistenceChangedPayload, ThreadCreatedPayload, ThreadDeletedPayload, ThreadWorktreeCleanupUpdatedPayload, @@ -324,6 +325,7 @@ export function projectEvent( unsettledAt: null, snoozedUntil: null, snoozedAt: null, + persistent: false, annotation: null, worktreeCleanup: null, deletedAt: null, @@ -395,6 +397,27 @@ export function projectEvent( })), ); + case "thread.persistence-changed": + return decodeForEvent( + ThreadPersistenceChangedPayload, + event.payload, + event.type, + "payload", + ).pipe( + Effect.map((payload) => ({ + ...nextBase, + threads: nextBase.threads.map((thread) => + thread.id === payload.threadId || thread.id === payload.replacedThreadId + ? { + ...thread, + persistent: thread.id === payload.persistentThreadId, + updatedAt: payload.updatedAt, + } + : thread, + ), + })), + ); + case "thread.settled": return decodeForEvent(ThreadSettledPayload, event.payload, event.type, "payload").pipe( Effect.map((payload) => ({ diff --git a/apps/server/src/persistence/Layers/ProjectionThreads.ts b/apps/server/src/persistence/Layers/ProjectionThreads.ts index eee6129db070..2dcf4e935490 100644 --- a/apps/server/src/persistence/Layers/ProjectionThreads.ts +++ b/apps/server/src/persistence/Layers/ProjectionThreads.ts @@ -62,6 +62,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { snoozed_until, snoozed_at, pinned_at, + persistent, pin_order_key, title_regeneration_request_id, title_regeneration_started_at, @@ -94,6 +95,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { ${row.snoozedUntil}, ${row.snoozedAt}, ${row.pinnedAt}, + ${row.persistent ?? 0}, ${row.pinOrderKey ?? null}, ${row.titleRegenerationRequestId ?? null}, ${row.titleRegenerationStartedAt ?? null}, @@ -126,6 +128,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { snoozed_until = excluded.snoozed_until, snoozed_at = excluded.snoozed_at, pinned_at = excluded.pinned_at, + persistent = excluded.persistent, pin_order_key = excluded.pin_order_key, title_regeneration_request_id = excluded.title_regeneration_request_id, title_regeneration_started_at = excluded.title_regeneration_started_at, @@ -165,6 +168,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", pinned_at AS "pinnedAt", + persistent, pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", @@ -206,6 +210,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", pinned_at AS "pinnedAt", + persistent, pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", @@ -255,6 +260,7 @@ const makeProjectionThreadRepository = Effect.gen(function* () { snoozed_until AS "snoozedUntil", snoozed_at AS "snoozedAt", pinned_at AS "pinnedAt", + persistent, pin_order_key AS "pinOrderKey", title_regeneration_request_id AS "titleRegenerationRequestId", title_regeneration_started_at AS "titleRegenerationStartedAt", diff --git a/apps/server/src/persistence/Migrations.ts b/apps/server/src/persistence/Migrations.ts index 5447d209ad68..87a75c7834e2 100644 --- a/apps/server/src/persistence/Migrations.ts +++ b/apps/server/src/persistence/Migrations.ts @@ -64,6 +64,7 @@ import Migration0048 from "./Migrations/048_ProjectionThreadWorktreeCleanup.ts"; import Migration0049 from "./Migrations/049_ProjectionThreadLinkedPullRequest.ts"; import Migration0050 from "./Migrations/050_ProjectionThreadsUnsettledAt.ts"; import Migration0051 from "./Migrations/051_ProjectionThreadMessageSource.ts"; +import Migration0052 from "./Migrations/052_ProjectionThreadsPersistent.ts"; /** * Migration loader with all migrations defined inline. @@ -127,6 +128,7 @@ export const migrationEntries = [ [49, "ProjectionThreadLinkedPullRequest", Migration0049], [50, "ProjectionThreadsUnsettledAt", Migration0050], [51, "ProjectionThreadMessageSource", Migration0051], + [52, "ProjectionThreadsPersistent", Migration0052], ] as const; export const migrationManifest = migrationEntries.map(([id, name]) => [id, name] as const); diff --git a/apps/server/src/persistence/Migrations/052_ProjectionThreadsPersistent.test.ts b/apps/server/src/persistence/Migrations/052_ProjectionThreadsPersistent.test.ts new file mode 100644 index 000000000000..cb6cd9c53e1f --- /dev/null +++ b/apps/server/src/persistence/Migrations/052_ProjectionThreadsPersistent.test.ts @@ -0,0 +1,27 @@ +import { assert, it } from "@effect/vitest"; +import * as Effect from "effect/Effect"; +import * as Layer from "effect/Layer"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +import { runMigrations } from "../Migrations.ts"; +import * as NodeSqliteClient from "../NodeSqliteClient.ts"; + +const layer = it.layer(Layer.mergeAll(NodeSqliteClient.layerMemory())); + +layer("052_ProjectionThreadsPersistent", (it) => { + it.effect("adds a false-by-default persistent marker", () => + Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + yield* runMigrations({ toMigrationInclusive: 51 }); + const executed = yield* runMigrations({ toMigrationInclusive: 52 }); + assert.deepStrictEqual(executed, [[52, "ProjectionThreadsPersistent"]]); + const columns = yield* sql<{ + readonly name: string; + readonly notnull: number; + readonly dflt_value: string | null; + }>`PRAGMA table_info(projection_threads)`; + const persistent = columns.find((column) => column.name === "persistent"); + assert.deepInclude(persistent, { notnull: 1, dflt_value: "0" }); + }), + ); +}); diff --git a/apps/server/src/persistence/Migrations/052_ProjectionThreadsPersistent.ts b/apps/server/src/persistence/Migrations/052_ProjectionThreadsPersistent.ts new file mode 100644 index 000000000000..095c64210790 --- /dev/null +++ b/apps/server/src/persistence/Migrations/052_ProjectionThreadsPersistent.ts @@ -0,0 +1,16 @@ +import * as Effect from "effect/Effect"; +import * as SqlClient from "effect/unstable/sql/SqlClient"; + +export default Effect.gen(function* () { + const sql = yield* SqlClient.SqlClient; + const columns = yield* sql<{ readonly name: string }>` + PRAGMA table_info(projection_threads) + `; + + if (!columns.some((column) => column.name === "persistent")) { + yield* sql` + ALTER TABLE projection_threads + ADD COLUMN persistent INTEGER NOT NULL DEFAULT 0 + `; + } +}); diff --git a/apps/server/src/persistence/Services/ProjectionThreads.ts b/apps/server/src/persistence/Services/ProjectionThreads.ts index 8cd10ae6811c..7b4d4ccd916a 100644 --- a/apps/server/src/persistence/Services/ProjectionThreads.ts +++ b/apps/server/src/persistence/Services/ProjectionThreads.ts @@ -48,6 +48,7 @@ export const ProjectionThread = Schema.Struct({ snoozedUntil: Schema.NullOr(IsoDateTime), snoozedAt: Schema.NullOr(IsoDateTime), pinnedAt: Schema.NullOr(IsoDateTime), + persistent: Schema.optional(NonNegativeInt), pinOrderKey: Schema.optional(Schema.NullOr(Schema.String)), titleRegenerationRequestId: Schema.optional(Schema.NullOr(CommandId)), titleRegenerationStartedAt: Schema.optional(Schema.NullOr(IsoDateTime)), diff --git a/apps/server/src/server.test.ts b/apps/server/src/server.test.ts index 6d29a9dffca5..e9f7bc664706 100644 --- a/apps/server/src/server.test.ts +++ b/apps/server/src/server.test.ts @@ -7516,6 +7516,182 @@ it.layer(NodeServices.layer)("server router seam", (it) => { }).pipe(Effect.provide(NodeHttpServer.layerTest), TestClock.withLive), ); + it.effect("subscribeShell refreshes both threads when persistence transfers", () => + Effect.gen(function* () { + const threadId = ThreadId.make("thread-persistent-new"); + const replacedThreadId = ThreadId.make("thread-persistent-old"); + const now = "2026-01-01T00:00:00.000Z"; + const event: OrchestrationEvent = { + sequence: 1, + eventId: EventId.make("event-persistence-transfer"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.persistence-changed", + payload: { + threadId, + persistentThreadId: threadId, + replacedThreadId, + updatedAt: now, + }, + }; + const trailingMessageEvent: OrchestrationEvent = { + sequence: 2, + eventId: EventId.make("event-after-persistence-transfer"), + aggregateKind: "thread", + aggregateId: threadId, + occurredAt: now, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.message-sent", + payload: {} as never, + }; + + assert.isFalse(isThreadDetailEvent(event)); + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + latestSequence: Effect.succeed(2), + readEvents: () => Stream.make(event, trailingMessageEvent), + }, + projectionSnapshotQuery: { + getThreadShellById: (requestedId) => + Effect.succeed( + Option.some( + makeDefaultOrchestrationThreadShell({ + id: requestedId, + persistent: requestedId === threadId, + }), + ), + ), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 0 }).pipe( + Stream.take(1), + Stream.runCollect, + ), + ), + ); + + const [first] = Array.from(items); + assert.equal(first?.kind, "thread-upserted"); + if (first?.kind !== "thread-upserted") return; + assert.equal(first.thread.id, threadId); + assert.equal(first.sequence, 2); + assert.isTrue(first.thread.persistent); + assert.deepEqual( + first.relatedThreads?.map((thread) => [thread.id, thread.persistent]), + [[replacedThreadId, false]], + ); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + + it.effect( + "subscribeShell preserves related refreshes when a transfer batch ends in removal", + () => + Effect.gen(function* () { + const removedThreadId = ThreadId.make("thread-persistent-removed"); + const replacedThreadId = ThreadId.make("thread-persistent-replaced"); + const now = "2026-01-01T00:00:00.000Z"; + const transferEvent = { + sequence: 1, + eventId: EventId.make("event-persistence-transfer-before-removal"), + aggregateKind: "thread", + aggregateId: removedThreadId, + occurredAt: now, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.persistence-changed", + payload: { + threadId: removedThreadId, + persistentThreadId: removedThreadId, + replacedThreadId, + updatedAt: now, + }, + } satisfies Extract; + const disableEvent = { + ...transferEvent, + sequence: 2, + eventId: EventId.make("event-persistence-disable-before-removal"), + payload: { + threadId: removedThreadId, + persistentThreadId: null, + replacedThreadId: removedThreadId, + updatedAt: now, + }, + } satisfies Extract; + const deletedEvent = { + sequence: 3, + eventId: EventId.make("event-delete-after-persistence-transfer"), + aggregateKind: "thread", + aggregateId: removedThreadId, + occurredAt: now, + commandId: null, + causationEventId: null, + correlationId: null, + metadata: {}, + type: "thread.deleted", + payload: { threadId: removedThreadId, deletedAt: now }, + } satisfies Extract; + + yield* buildAppUnderTest({ + layers: { + orchestrationEngine: { + latestSequence: Effect.succeed(3), + readEvents: () => Stream.make(transferEvent, disableEvent, deletedEvent), + }, + projectionSnapshotQuery: { + getThreadShellById: (requestedId) => + Effect.succeed( + requestedId === removedThreadId + ? Option.none() + : Option.some( + makeDefaultOrchestrationThreadShell({ + id: requestedId, + persistent: false, + }), + ), + ), + }, + }, + }); + + const wsUrl = yield* getWsServerUrl("/ws"); + const items = yield* Effect.scoped( + withWsRpcClient(wsUrl, (client) => + client[ORCHESTRATION_WS_METHODS.subscribeShell]({ afterSequence: 0 }).pipe( + Stream.take(1), + Stream.runCollect, + ), + ), + ); + + const [first] = Array.from(items); + assert.equal(first?.kind, "thread-removed"); + if (first?.kind !== "thread-removed") return; + assert.equal(first.threadId, removedThreadId); + assert.equal(first.sequence, 3); + assert.deepEqual( + first.relatedThreads?.map((thread) => [thread.id, thread.persistent]), + [[replacedThreadId, false]], + ); + }).pipe(Effect.provide(NodeHttpServer.layerTest)), + ); + it.effect("subscribeShell coalescing still emits a removal for a deleted thread", () => Effect.gen(function* () { const goneThreadId = ThreadId.make("thread-gone"); diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts index 1ca52145b404..e9f816ae4468 100644 --- a/apps/server/src/ws.ts +++ b/apps/server/src/ws.ts @@ -694,6 +694,7 @@ const makeWsRpcLayer = ( const toShellStreamEvent = ( event: OrchestrationEvent, + relatedThreadIds: ReadonlyArray = [], ): Effect.Effect, never, never> => { switch (event.type) { case "project.created": @@ -708,22 +709,54 @@ const makeWsRpcLayer = ( }), ); case "thread.archived": - return Effect.succeed( - Option.some({ - kind: "thread-removed" as const, - sequence: event.sequence, - threadId: event.payload.threadId, - }), - ); + return relatedThreadIds.length === 0 + ? Effect.succeed( + Option.some({ + kind: "thread-removed" as const, + sequence: event.sequence, + threadId: event.payload.threadId, + }), + ) + : threadUpsertOrRemoveWithRelated( + event.payload.threadId, + relatedThreadIds, + event.sequence, + ); case "thread.deleted": - return threadUpsertOrRemove(event.payload.threadId, event.sequence); + return relatedThreadIds.length === 0 + ? threadUpsertOrRemove(event.payload.threadId, event.sequence) + : threadUpsertOrRemoveWithRelated( + event.payload.threadId, + relatedThreadIds, + event.sequence, + ); case "thread.unarchived": - return threadUpsertOrRemove(event.payload.threadId, event.sequence); + return relatedThreadIds.length === 0 + ? threadUpsertOrRemove(event.payload.threadId, event.sequence) + : threadUpsertOrRemoveWithRelated( + event.payload.threadId, + relatedThreadIds, + event.sequence, + ); + case "thread.persistence-changed": + return relatedThreadIds.length === 0 + ? threadUpsertOrRemove(event.payload.threadId, event.sequence) + : threadUpsertOrRemoveWithRelated( + event.payload.threadId, + relatedThreadIds, + event.sequence, + ); default: if (event.aggregateKind !== "thread") { return Effect.succeed(Option.none()); } - return threadUpsertOrRemove(ThreadId.make(event.aggregateId), event.sequence); + return relatedThreadIds.length === 0 + ? threadUpsertOrRemove(ThreadId.make(event.aggregateId), event.sequence) + : threadUpsertOrRemoveWithRelated( + ThreadId.make(event.aggregateId), + relatedThreadIds, + event.sequence, + ); } }; @@ -818,6 +851,30 @@ const makeWsRpcLayer = ( ), ); + const threadUpsertOrRemoveWithRelated = ( + threadId: ThreadId, + relatedThreadIds: ReadonlyArray, + sequence: number, + ): Effect.Effect, never, never> => + Effect.gen(function* () { + const primary = yield* threadUpsertOrRemove(threadId, sequence); + if (Option.isNone(primary)) { + return primary; + } + const related = yield* Effect.forEach( + [...new Set(relatedThreadIds)].filter((relatedId) => relatedId !== threadId), + (relatedId) => threadUpsertOrRemove(relatedId, sequence), + { concurrency: SHELL_REFETCH_CONCURRENCY }, + ); + const relatedThreads = related.flatMap((item) => + Option.isSome(item) && item.value.kind === "thread-upserted" ? [item.value.thread] : [], + ); + return Option.some({ + ...primary.value, + ...(relatedThreads.length > 0 ? { relatedThreads } : {}), + }); + }); + // Turn a batch of domain events into shell stream items, coalescing by // aggregate first. `toShellStreamEvent` re-reads the *current* projected // shell for an aggregate, so within a batch only the latest event per @@ -839,15 +896,35 @@ const makeWsRpcLayer = ( return []; } const latestByAggregate = new Map(); + const relatedThreadIdsByAggregate = new Map>(); for (const event of events) { - latestByAggregate.set(`${event.aggregateKind}:${event.aggregateId}`, event); + const aggregateKey = `${event.aggregateKind}:${event.aggregateId}`; + latestByAggregate.set(aggregateKey, event); + if ( + event.type === "thread.persistence-changed" && + event.payload.replacedThreadId !== null && + event.payload.replacedThreadId !== event.payload.threadId + ) { + const related = relatedThreadIdsByAggregate.get(aggregateKey) ?? new Set(); + related.add(event.payload.replacedThreadId); + relatedThreadIdsByAggregate.set(aggregateKey, related); + } } const survivors = Array.from(latestByAggregate.values()).sort( (left, right) => left.sequence - right.sequence, ); - const shellEvents = yield* Effect.forEach(survivors, toShellStreamEvent, { - concurrency: SHELL_REFETCH_CONCURRENCY, - }); + const shellEvents = yield* Effect.forEach( + survivors, + (event) => + toShellStreamEvent( + event, + Array.from( + relatedThreadIdsByAggregate.get(`${event.aggregateKind}:${event.aggregateId}`) ?? + [], + ), + ), + { concurrency: SHELL_REFETCH_CONCURRENCY }, + ); return shellEvents.flatMap((option) => (Option.isSome(option) ? [option.value] : [])); }); diff --git a/apps/web/src/components/LegacySidebar.tsx b/apps/web/src/components/LegacySidebar.tsx index d40cda858113..ee0b9077d5a6 100644 --- a/apps/web/src/components/LegacySidebar.tsx +++ b/apps/web/src/components/LegacySidebar.tsx @@ -5,6 +5,7 @@ import { FolderPlusIcon, Globe2Icon, LoaderIcon, + MessageSquareLockIcon, SearchIcon, SquarePenIcon, TerminalIcon, @@ -93,6 +94,7 @@ import { isMacPlatform } from "../lib/utils"; import { readThreadShell, readEnvironmentSupportsThreadAnnotations, + readEnvironmentSupportsPersistence, useProject, useProjects, useThreadShells, @@ -160,6 +162,11 @@ import { shouldToastDesktopUpdateActionResult, } from "./desktopUpdate.logic"; import { showDesktopUpdateDownloadedToast } from "./desktopUpdate.toast"; +import { + legacyThreadPersistenceAction, + protectLegacyThreadActions, +} from "./legacyThreadPersistence.logic"; +import { projectsContainPersistentThread } from "./projectPersistence.logic"; import { Alert, AlertAction, AlertDescription, AlertTitle } from "./ui/alert"; import { Button } from "./ui/button"; import { @@ -198,6 +205,7 @@ import { openCommandPalette } from "../commandPaletteBus"; import { archiveSelectedThreadEntries, buildMultiSelectThreadContextMenuItems, + collectUnprotectedBulkThreadEntries, getSidebarThreadIdsToPrewarm, resolveAdjacentThreadId, isContextMenuPointerDown, @@ -804,28 +812,31 @@ export const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThr event.preventDefault(); event.stopPropagation(); clearConfirmingArchive(); + if (thread.persistent) return; void attemptArchiveThread(threadRef); }, - [attemptArchiveThread, clearConfirmingArchive, threadRef], + [attemptArchiveThread, clearConfirmingArchive, thread.persistent, threadRef], ); const handleStartArchiveConfirmation = useCallback( (event: React.MouseEvent) => { event.preventDefault(); event.stopPropagation(); + if (thread.persistent) return; setConfirmingArchiveThreadKey(threadKey); requestAnimationFrame(() => { confirmArchiveButtonRefs.current.get(threadKey)?.focus(); }); }, - [confirmArchiveButtonRefs, setConfirmingArchiveThreadKey, threadKey], + [confirmArchiveButtonRefs, setConfirmingArchiveThreadKey, thread.persistent, threadKey], ); const handleArchiveImmediateClick = useCallback( (event: React.MouseEvent) => { event.preventDefault(); event.stopPropagation(); + if (thread.persistent) return; void attemptArchiveThread(threadRef); }, - [attemptArchiveThread, threadRef], + [attemptArchiveThread, thread.persistent, threadRef], ); const threadDetailsTooltipHandle = useMemo(() => TooltipCreateHandle(), []); const rowButtonRender = useMemo( @@ -943,12 +954,17 @@ export const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThr onDoubleClick={handleRenameInputClick} /> ) : ( - - {thread.title} - + <> + {thread.persistent ? ( + + ) : null} + + {thread.title} + + )}
@@ -1041,7 +1057,7 @@ export const SidebarThreadRow = memo(function SidebarThreadRow(props: SidebarThr ) : null} - {isConfirmingArchive ? ( + {isConfirmingArchive && !thread.persistent ? ( - ) : !isThreadRunning && cleanup === null ? ( + ) : !thread.persistent && !isThreadRunning && cleanup === null ? ( appSettingsConfirmThreadArchive ? (
} /> diff --git a/apps/web/src/components/threadActionMenu.logic.test.ts b/apps/web/src/components/threadActionMenu.logic.test.ts index e56515a4ecfa..605e8c018186 100644 --- a/apps/web/src/components/threadActionMenu.logic.test.ts +++ b/apps/web/src/components/threadActionMenu.logic.test.ts @@ -5,13 +5,20 @@ import { buildThreadActionMenuItems, type ThreadActionMenuState } from "./thread const baseState: ThreadActionMenuState = { branch: null, isPinned: false, + isPersistent: false, isSettled: false, isSnoozed: false, canSnoozeNow: true, isRegeneratingTitle: false, isRunning: false, hasRunningAction: false, - supports: { settlement: true, snooze: true, pinning: true, titleRegeneration: true }, + supports: { + settlement: true, + snooze: true, + pinning: true, + persistence: true, + titleRegeneration: true, + }, snoozePresets: [ { id: "hour", label: "In 1 hour", whenLabel: "3:00 PM", snoozedUntil: "2026-08-07T15:00:00Z" }, ], @@ -32,7 +39,13 @@ describe("buildThreadActionMenuItems", () => { expect( ids({ ...baseState, - supports: { settlement: false, snooze: false, pinning: false, titleRegeneration: false }, + supports: { + settlement: false, + snooze: false, + pinning: false, + persistence: false, + titleRegeneration: false, + }, }), ).toEqual(["rename", "mark-unread", "copy", "archive", "delete"]); }); @@ -90,7 +103,13 @@ describe("buildThreadActionMenuItems", () => { expect( ids({ ...baseState, - supports: { settlement: false, snooze: false, pinning: false, titleRegeneration: false }, + supports: { + settlement: false, + snooze: false, + pinning: false, + persistence: false, + titleRegeneration: false, + }, }), ).toContain("archive"); }); @@ -101,4 +120,15 @@ describe("buildThreadActionMenuItems", () => { ); expect(archiveItem?.disabled).toBe(true); }); + + it("replaces the mark action and blocks archive and delete for the persistent thread", () => { + const items = buildThreadActionMenuItems({ ...baseState, isPersistent: true }); + expect(items).toContainEqual( + expect.objectContaining({ id: "disable-persistence", label: "Disable persistent thread" }), + ); + expect(items.find((item) => item.id === "archive")?.disabled).toBe(true); + expect(items.find((item) => item.id === "archive")?.label).toContain("disable persistence"); + expect(items.find((item) => item.id === "delete")?.disabled).toBe(true); + expect(items.find((item) => item.id === "delete")?.label).toContain("disable persistence"); + }); }); diff --git a/apps/web/src/components/threadActionMenu.logic.ts b/apps/web/src/components/threadActionMenu.logic.ts index 479c9a13bf80..36b228457a7f 100644 --- a/apps/web/src/components/threadActionMenu.logic.ts +++ b/apps/web/src/components/threadActionMenu.logic.ts @@ -10,6 +10,8 @@ export type ThreadActionMenuId = | "new-thread-on-branch" | "pin" | "unpin" + | "mark-persistent" + | "disable-persistence" | "settle" | "unsettle" | "snooze" @@ -29,6 +31,7 @@ export type ThreadActionMenuId = export interface ThreadActionMenuState { readonly branch: string | null; readonly isPinned: boolean; + readonly isPersistent: boolean; readonly isSettled: boolean; readonly isSnoozed: boolean; readonly canSnoozeNow: boolean; @@ -40,6 +43,7 @@ export interface ThreadActionMenuState { readonly settlement: boolean; readonly snooze: boolean; readonly pinning: boolean; + readonly persistence: boolean; readonly titleRegeneration: boolean; }; readonly snoozePresets: ReadonlyArray; @@ -70,6 +74,21 @@ export function buildThreadActionMenuItems( : { id: "pin" as const, label: "Pin thread", icon: "pin" }, ] : []), + ...(state.supports.persistence + ? [ + state.isPersistent + ? { + id: "disable-persistence" as const, + label: "Disable persistent thread", + icon: "message-square-lock", + } + : { + id: "mark-persistent" as const, + label: "Mark as persistent thread", + icon: "message-square-lock", + }, + ] + : []), // Both lifecycle actions stay available on pinned threads: settling // clears the pin ("done" beats "keep on top"), and snoozing hides the // card until wake with the pin intact. @@ -131,16 +150,17 @@ export function buildThreadActionMenuItems( // styling. { id: "archive", - label: "Archive thread", + label: state.isPersistent ? "Archive thread (disable persistence first)" : "Archive thread", icon: "archive", - disabled: state.isRunning, + disabled: state.isRunning || state.isPersistent, separatorBefore: true, }, { id: "delete", - label: "Delete", + label: state.isPersistent ? "Delete (disable persistence first)" : "Delete", destructive: true, icon: "trash", + disabled: state.isPersistent, }, ]; } diff --git a/apps/web/src/contextMenuFallback.ts b/apps/web/src/contextMenuFallback.ts index ce8b8950a8b9..ec55ded1951f 100644 --- a/apps/web/src/contextMenuFallback.ts +++ b/apps/web/src/contextMenuFallback.ts @@ -70,6 +70,16 @@ const ICON_PATHS: Record confirmAndUnpinThread(threadRef)); return; } + case "mark-persistent": + await reportFailure("Failed to mark persistent thread", () => + setThreadPersistence(threadRef, true), + ); + return; + case "disable-persistence": + await reportFailure("Failed to disable persistent thread", () => + setThreadPersistence(threadRef, false), + ); + return; case "rename": onStartRename(); return; @@ -342,6 +356,7 @@ export function useThreadActionMenu(input: { pinThread, projectCwd, settleThread, + setThreadPersistence, snoozeThread, threadRef, timestampFormat, diff --git a/apps/web/src/hooks/useThreadActions.ts b/apps/web/src/hooks/useThreadActions.ts index 8f912c46bc56..a9e0655d3731 100644 --- a/apps/web/src/hooks/useThreadActions.ts +++ b/apps/web/src/hooks/useThreadActions.ts @@ -228,6 +228,9 @@ export function useThreadActions() { const unarchiveThreadMutation = useAtomCommand(threadEnvironment.unarchive, { reportFailure: false, }); + const setThreadPersistenceMutation = useAtomCommand(threadEnvironment.setPersistence, { + reportFailure: false, + }); const deleteThreadMutation = useAtomCommand(threadEnvironment.delete, { reportFailure: false, }); @@ -355,6 +358,15 @@ export function useThreadActions() { [unarchiveThreadMutation], ); + const setThreadPersistence = useCallback( + (target: ScopedThreadRef, persistent: boolean) => + setThreadPersistenceMutation({ + environmentId: target.environmentId, + input: { threadId: target.threadId, persistent }, + }), + [setThreadPersistenceMutation], + ); + const deleteThread = useCallback( async (target: ScopedThreadRef, opts: DeleteThreadOptions = {}) => { const resolved = resolveThreadTargetWithArchivedFallback( @@ -842,6 +854,7 @@ export function useThreadActions() { () => ({ archiveThread, unarchiveThread, + setThreadPersistence, deleteThread, confirmAndDeleteThread, settleThread, @@ -863,6 +876,7 @@ export function useThreadActions() { settleThread, snoozeThread, unarchiveThread, + setThreadPersistence, unpinThread, unsettleThread, unsnoozeThread, diff --git a/apps/web/src/state/entities.ts b/apps/web/src/state/entities.ts index 6b213711d512..a84d952cf08b 100644 --- a/apps/web/src/state/entities.ts +++ b/apps/web/src/state/entities.ts @@ -250,6 +250,13 @@ export function readEnvironmentSupportsPinning(environmentId: EnvironmentId): bo ); } +export function readEnvironmentSupportsPersistence(environmentId: EnvironmentId): boolean { + return ( + appAtomRegistry.get(environmentServerConfigsAtom).get(environmentId)?.environment.capabilities + .threadPersistence === true + ); +} + /** Whether the environment's server understands thread title regeneration. Same version-skew contract as settlement. */ export function readEnvironmentSupportsTitleRegeneration(environmentId: EnvironmentId): boolean { diff --git a/docs/lastcode/nightly-workflow.md b/docs/lastcode/nightly-workflow.md index 05df6c5bc3e6..42e7d1f6e54f 100644 --- a/docs/lastcode/nightly-workflow.md +++ b/docs/lastcode/nightly-workflow.md @@ -237,7 +237,9 @@ pnpm lastcode:checkpoint:service install \ ``` For unattended recovery, dedicate one durable LastCode thread to checkpoint -maintenance and configure its thread ID once: +maintenance. Open that thread's context menu (right-click on desktop/web or +long-press on mobile), choose **Mark as persistent thread**, then configure its +thread ID once: ```bash pnpm lastcode:checkpoint:service install \ @@ -246,7 +248,14 @@ pnpm lastcode:checkpoint:service install \ ``` The service reuses that thread instead of creating a new thread per failure. A -standalone supervisor covers fetch, checkout, dependency setup, and checkpoint +message-square-lock marker and italic title identify it in LastCode. The server +blocks archive and permanent deletion while the thread is persistent. Marking +another thread persistent atomically transfers the safeguard. A project that +contains the persistent thread cannot be removed either, because project removal +would delete its threads. Choose **Disable persistent thread** when recovery +delivery no longer needs protection. + +A standalone supervisor covers fetch, checkout, dependency setup, and checkpoint execution; every run writes terminal state to `~/.lastcode/automation/checkpoint-service-state.json`. A failed alert remains pending until LastCode accepts it, and the same incident is not sent again after @@ -321,6 +330,15 @@ lastcode-checkpoints -n 20 lastcode-checkpoints --verbose ``` +The dashboard compares the supervisor's configured recovery thread ID with +LastCode's authoritative thread list. If the ID is missing or no longer marked +persistent, it prints the mismatch and repair guidance. After designating the +replacement in LastCode, repair the supervisor configuration atomically with: + +```bash +lastcode-checkpoints --repair-persistent-thread +``` + The dashboard shows success or failure, upstream nightly, number of downstream commits replayed, finish time, duration, checkpoint commit, promotion to `lastcode/main`, and whether a local build tag exists. Built LastCode revisions diff --git a/packages/client-runtime/src/operations/commands.ts b/packages/client-runtime/src/operations/commands.ts index 2b6f5bd8f939..6a17aab4d7e0 100644 --- a/packages/client-runtime/src/operations/commands.ts +++ b/packages/client-runtime/src/operations/commands.ts @@ -37,6 +37,7 @@ export type RetryThreadWorktreeCleanupInput = CommandInput<"thread.worktree-clea export type AbandonThreadWorktreeCleanupInput = CommandInput<"thread.worktree-cleanup.abandon">; export type ArchiveThreadInput = CommandInput<"thread.archive">; export type UnarchiveThreadInput = CommandInput<"thread.unarchive">; +export type SetThreadPersistenceInput = CommandInput<"thread.persistence.set">; export type SettleThreadInput = CommandInput<"thread.settle">; export type UnsettleThreadInput = CommandInput<"thread.unsettle">; export type SnoozeThreadInput = CommandInput<"thread.snooze">; @@ -186,6 +187,16 @@ export const unarchiveThread: (input: UnarchiveThreadInput) => CommandEffect = E }); }); +export const setThreadPersistence: (input: SetThreadPersistenceInput) => CommandEffect = Effect.fn( + "EnvironmentCommands.setThreadPersistence", +)(function* (input) { + return yield* dispatch({ + ...input, + type: "thread.persistence.set", + commandId: yield* commandId(input), + }); +}); + export const settleThread: (input: SettleThreadInput) => CommandEffect = Effect.fn( "EnvironmentCommands.settleThread", )(function* (input) { diff --git a/packages/client-runtime/src/state/entities.test.ts b/packages/client-runtime/src/state/entities.test.ts index d3bb6680208a..b3ee43042676 100644 --- a/packages/client-runtime/src/state/entities.test.ts +++ b/packages/client-runtime/src/state/entities.test.ts @@ -203,6 +203,7 @@ describe("environment entity projections", () => { ...THREAD_SHELL, environmentId: ENVIRONMENT_ID, title: "Cached thread", + persistent: true, branch: "stale-branch", worktreePath: "/repo/stale-worktree", deletedAt: null, @@ -215,6 +216,7 @@ describe("environment entity projections", () => { ...THREAD_SHELL, environmentId: ENVIRONMENT_ID, title: "Current thread", + persistent: false, branch: "current-branch", worktreePath: "/repo/current-worktree", }; @@ -223,6 +225,7 @@ describe("environment entity projections", () => { expect(merged).toMatchObject({ title: "Current thread", + persistent: false, branch: "current-branch", worktreePath: "/repo/current-worktree", }); diff --git a/packages/client-runtime/src/state/shellReducer.test.ts b/packages/client-runtime/src/state/shellReducer.test.ts index fdccc4c47dd8..b0c8b4dbc9f1 100644 --- a/packages/client-runtime/src/state/shellReducer.test.ts +++ b/packages/client-runtime/src/state/shellReducer.test.ts @@ -155,6 +155,32 @@ describe("applyShellStreamEvent", () => { expect(next.threads).toHaveLength(1); expect(next.threads[0]?.title).toBe("Updated Thread"); }); + + it("updates related threads in the same sequence", () => { + const replacedThread = { + ...stubThread, + id: ThreadId.make("thread-2"), + title: "Previously persistent", + persistent: true, + }; + const snapshotWithThreads: OrchestrationShellSnapshot = { + ...baseSnapshot, + threads: [stubThread, replacedThread], + }; + + const next = applyShellStreamEvent(snapshotWithThreads, { + kind: "thread-upserted", + sequence: 6, + thread: { ...stubThread, persistent: true }, + relatedThreads: [{ ...replacedThread, persistent: false }], + }); + + expect(next.threads.map((thread) => [thread.id, thread.persistent])).toEqual([ + ["thread-1", true], + ["thread-2", false], + ]); + expect(next.snapshotSequence).toBe(6); + }); }); describe("thread-removed", () => { @@ -175,6 +201,32 @@ describe("applyShellStreamEvent", () => { expect(next.threads).toHaveLength(0); expect(next.snapshotSequence).toBe(6); }); + + it("refreshes related threads in the removal sequence", () => { + const removedThread = { ...stubThread, persistent: true }; + const relatedThread = { + ...stubThread, + id: ThreadId.make("thread-2"), + title: "Previously persistent", + persistent: true, + }; + const snapshotWithThreads: OrchestrationShellSnapshot = { + ...baseSnapshot, + threads: [removedThread, relatedThread], + }; + + const next = applyShellStreamEvent(snapshotWithThreads, { + kind: "thread-removed", + sequence: 7, + threadId: removedThread.id, + relatedThreads: [{ ...relatedThread, persistent: false }], + }); + + expect(next.threads.map((thread) => [thread.id, thread.persistent])).toEqual([ + ["thread-2", false], + ]); + expect(next.snapshotSequence).toBe(7); + }); }); it("returns original snapshot for unrecognized event kinds", () => { diff --git a/packages/client-runtime/src/state/shellReducer.ts b/packages/client-runtime/src/state/shellReducer.ts index 3d3b22a1289f..0382a1026b64 100644 --- a/packages/client-runtime/src/state/shellReducer.ts +++ b/packages/client-runtime/src/state/shellReducer.ts @@ -1,5 +1,22 @@ import * as Arr from "effect/Array"; -import type { OrchestrationShellSnapshot, OrchestrationShellStreamEvent } from "@t3tools/contracts"; +import type { + OrchestrationShellSnapshot, + OrchestrationShellStreamEvent, + OrchestrationThreadShell, +} from "@t3tools/contracts"; + +function upsertThreadShells( + threads: ReadonlyArray, + updates: ReadonlyArray, +): ReadonlyArray { + return updates.reduce( + (current, nextThread) => + current.some((thread) => thread.id === nextThread.id) + ? Arr.map(current, (thread) => (thread.id === nextThread.id ? nextThread : thread)) + : Arr.append(current, nextThread), + threads, + ); +} /** * Reduce a single shell stream event into an existing snapshot, returning a new @@ -29,17 +46,20 @@ export function applyShellStreamEvent( snapshotSequence: event.sequence, }; case "thread-upserted": { - const threads = snapshot.threads.some((t) => t.id === event.thread.id) - ? Arr.map(snapshot.threads, (t) => (t.id === event.thread.id ? event.thread : t)) - : Arr.append(snapshot.threads, event.thread); + const threads = upsertThreadShells(snapshot.threads, [ + event.thread, + ...(event.relatedThreads ?? []), + ]); return { ...snapshot, threads, snapshotSequence: event.sequence }; } - case "thread-removed": + case "thread-removed": { + const refreshedThreads = upsertThreadShells(snapshot.threads, event.relatedThreads ?? []); return { ...snapshot, - threads: Arr.filter(snapshot.threads, (t) => t.id !== event.threadId), + threads: Arr.filter(refreshedThreads, (thread) => thread.id !== event.threadId), snapshotSequence: event.sequence, }; + } default: return snapshot; } diff --git a/packages/client-runtime/src/state/threadCommands.ts b/packages/client-runtime/src/state/threadCommands.ts index 18f23ce80429..b18737607375 100644 --- a/packages/client-runtime/src/state/threadCommands.ts +++ b/packages/client-runtime/src/state/threadCommands.ts @@ -29,6 +29,7 @@ import { type StartThreadTurnInput, type StopThreadSessionInput, type UnarchiveThreadInput, + type SetThreadPersistenceInput, type UnpinThreadInput, type UnsettleThreadInput, type UnsnoozeThreadInput, @@ -54,6 +55,7 @@ import { startThreadTurn, stopThreadSession, unarchiveThread, + setThreadPersistence, unpinThread, unsettleThread, unsnoozeThread, @@ -83,6 +85,7 @@ export type { StartThreadTurnInput, StopThreadSessionInput, UnarchiveThreadInput, + SetThreadPersistenceInput, UnpinThreadInput, UnsettleThreadInput, UnsnoozeThreadInput, @@ -135,6 +138,12 @@ export function createThreadEnvironmentAtoms( scheduler, concurrency, }), + setPersistence: createEnvironmentCommand(runtime, { + label: "environment-data:commands:thread:persistence:set", + execute: (input: SetThreadPersistenceInput) => setThreadPersistence(input), + scheduler, + concurrency, + }), settle: createEnvironmentCommand(runtime, { label: "environment-data:commands:thread:settle", execute: (input: SettleThreadInput) => settleThread(input), diff --git a/packages/client-runtime/src/state/threadDetail.ts b/packages/client-runtime/src/state/threadDetail.ts index 46d1f91722a7..34942020ccd7 100644 --- a/packages/client-runtime/src/state/threadDetail.ts +++ b/packages/client-runtime/src/state/threadDetail.ts @@ -62,6 +62,7 @@ export function mergeEnvironmentThread( snoozedUntil: shell.snoozedUntil, snoozedAt: shell.snoozedAt, pinnedAt: shell.pinnedAt, + ...(shell.persistent === undefined ? {} : { persistent: shell.persistent }), pinOrderKey: shell.pinOrderKey, annotation: shell.annotation, session: shell.session, diff --git a/packages/client-runtime/src/state/threadReducer.ts b/packages/client-runtime/src/state/threadReducer.ts index c4939da298c5..4c6dc501e2de 100644 --- a/packages/client-runtime/src/state/threadReducer.ts +++ b/packages/client-runtime/src/state/threadReducer.ts @@ -95,6 +95,7 @@ export function applyThreadDetailEvent( unsettledAt: null, snoozedUntil: null, snoozedAt: null, + persistent: false, annotation: null, deletedAt: null, messages: [], diff --git a/packages/contracts/src/environment.ts b/packages/contracts/src/environment.ts index 8ac0d30f9a04..06761b0173bd 100644 --- a/packages/contracts/src/environment.ts +++ b/packages/contracts/src/environment.ts @@ -74,6 +74,8 @@ export const ExecutionEnvironmentCapabilities = Schema.Struct({ /** Server understands thread.pin / thread.unpin commands. Same version-skew contract as threadSettlement. */ threadPinning: Schema.optionalKey(Schema.Boolean), + /** Server understands thread.persistence.set and enforces archive/delete protection. */ + threadPersistence: Schema.optionalKey(Schema.Boolean), /** Server understands thread.pin.reorder (and orderKey on thread.pin). Same version-skew contract as threadSettlement. */ threadPinReorder: Schema.optionalKey(Schema.Boolean), diff --git a/packages/contracts/src/orchestration.ts b/packages/contracts/src/orchestration.ts index b7bf5d915501..15c1740aebdd 100644 --- a/packages/contracts/src/orchestration.ts +++ b/packages/contracts/src/orchestration.ts @@ -733,6 +733,8 @@ export const OrchestrationThread = Schema.Struct({ // threads remain in their respective shelves even when pinned. // Optional so payloads from pre-pinning servers still decode. pinnedAt: Schema.optional(Schema.NullOr(IsoDateTime)), + // Optional on the wire so pre-safeguard clients and cached snapshots decode it as false. + persistent: Schema.optional(Schema.Boolean), // Fractional index for user-arranged pinned order. Keyed threads sort by // string comparison ahead of keyless ones (which keep creation order), so // servers never need each other's threads to agree on the merged list. @@ -806,6 +808,7 @@ export const OrchestrationThreadShell = Schema.Struct({ snoozedUntil: Schema.optional(Schema.NullOr(IsoDateTime)), snoozedAt: Schema.optional(Schema.NullOr(IsoDateTime)), pinnedAt: Schema.optional(Schema.NullOr(IsoDateTime)), + persistent: Schema.optional(Schema.Boolean), pinOrderKey: Schema.optional(Schema.NullOr(TrimmedNonEmptyString)), titleRegeneration: Schema.optional(Schema.NullOr(ThreadTitleRegeneration)), annotation: Schema.optional(Schema.NullOr(ThreadAnnotation)), @@ -867,11 +870,13 @@ export const OrchestrationShellStreamEvent = Schema.Union([ kind: Schema.Literal("thread-upserted"), sequence: NonNegativeInt, thread: OrchestrationThreadShell, + relatedThreads: Schema.optional(Schema.Array(OrchestrationThreadShell)), }), Schema.Struct({ kind: Schema.Literal("thread-removed"), sequence: NonNegativeInt, threadId: ThreadId, + relatedThreads: Schema.optional(Schema.Array(OrchestrationThreadShell)), }), ]); export type OrchestrationShellStreamEvent = typeof OrchestrationShellStreamEvent.Type; @@ -1070,6 +1075,13 @@ const ThreadUnarchiveCommand = Schema.Struct({ threadId: ThreadId, }); +const ThreadPersistenceSetCommand = Schema.Struct({ + type: Schema.Literal("thread.persistence.set"), + commandId: CommandId, + threadId: ThreadId, + persistent: Schema.Boolean, +}); + const ThreadSettleCommand = Schema.Struct({ type: Schema.Literal("thread.settle"), commandId: CommandId, @@ -1318,6 +1330,7 @@ const DispatchableClientOrchestrationCommand = Schema.Union([ ThreadWorktreeCleanupAbandonCommand, ThreadArchiveCommand, ThreadUnarchiveCommand, + ThreadPersistenceSetCommand, ThreadSettleCommand, ThreadUnsettleCommand, ThreadSnoozeCommand, @@ -1352,6 +1365,7 @@ export const ClientOrchestrationCommand = Schema.Union([ ThreadWorktreeCleanupAbandonCommand, ThreadArchiveCommand, ThreadUnarchiveCommand, + ThreadPersistenceSetCommand, ThreadSettleCommand, ThreadUnsettleCommand, ThreadSnoozeCommand, @@ -1511,6 +1525,7 @@ export const OrchestrationEventType = Schema.Literals([ "thread.worktree-cleanup-updated", "thread.archived", "thread.unarchived", + "thread.persistence-changed", "thread.settled", "thread.unsettled", "thread.snoozed", @@ -1613,6 +1628,13 @@ export const ThreadUnarchivedPayload = Schema.Struct({ updatedAt: IsoDateTime, }); +export const ThreadPersistenceChangedPayload = Schema.Struct({ + threadId: ThreadId, + persistentThreadId: Schema.NullOr(ThreadId), + replacedThreadId: Schema.NullOr(ThreadId), + updatedAt: IsoDateTime, +}); + export const ThreadSettledPayload = Schema.Struct({ threadId: ThreadId, settledAt: IsoDateTime, @@ -1874,6 +1896,11 @@ export const OrchestrationEvent = Schema.Union([ type: Schema.Literal("thread.unarchived"), payload: ThreadUnarchivedPayload, }), + Schema.Struct({ + ...EventBaseFields, + type: Schema.Literal("thread.persistence-changed"), + payload: ThreadPersistenceChangedPayload, + }), Schema.Struct({ ...EventBaseFields, type: Schema.Literal("thread.settled"), diff --git a/scripts/lastcode-checkpoints.mjs b/scripts/lastcode-checkpoints.mjs index d4d334a3a681..0ce46171862b 100644 --- a/scripts/lastcode-checkpoints.mjs +++ b/scripts/lastcode-checkpoints.mjs @@ -55,12 +55,14 @@ function splitLines(value) { export function parseOptions(argv) { let count = DEFAULT_COUNT; let install = false; + let repairPersistentThread = false; let repoRoot; let verbose = false; for (let index = 0; index < argv.length; index += 1) { const arg = argv[index]; if (arg === "--") continue; if (arg === "--install") install = true; + else if (arg === "--repair-persistent-thread") repairPersistentThread = true; else if (arg === "-v" || arg === "--verbose") verbose = true; else if (arg === "-n" || arg === "--count" || arg === "--repo") { const value = argv[index + 1]; @@ -74,12 +76,112 @@ export function parseOptions(argv) { } index += 1; } else if (arg === "-h" || arg === "--help") { - return { help: true, count, install, repoRoot, verbose }; + return { help: true, count, install, repairPersistentThread, repoRoot, verbose }; } else { throw new Error(`Unknown argument '${arg}'.`); } } - return { help: false, count, install, repoRoot, verbose }; + return { help: false, count, install, repairPersistentThread, repoRoot, verbose }; +} + +export function readCheckpointSupervisorConfig(home) { + const configPath = NodePath.join(home, ".lastcode", "automation", "checkpoint-supervisor.json"); + if (!NodeFS.existsSync(configPath)) return { configPath, config: null }; + try { + const config = JSON.parse(NodeFS.readFileSync(configPath, "utf8")); + return { configPath, config: config?.schemaVersion === 1 ? config : null }; + } catch { + return { configPath, config: null }; + } +} + +export function persistentThreadRepairStatus(config, threads) { + const configuredId = + typeof config?.recoveryThreadId === "string" ? config.recoveryThreadId : undefined; + if (!configuredId) return { kind: "disabled" }; + if (threads === null) return { kind: "unavailable", configuredId }; + const configuredThread = threads.find((thread) => thread.threadId === configuredId); + if (configuredThread?.persistent === true) return { kind: "healthy", configuredId }; + const replacement = threads.find((thread) => thread.persistent === true); + return { + kind: configuredThread ? "not-persistent" : "stale", + configuredId, + ...(replacement + ? { replacementId: replacement.threadId, replacementTitle: replacement.title } + : {}), + }; +} + +function readLastCodeThreads(home) { + const toolPath = NodePath.join(home, ".lastcode", "userdata", "bin", "lastcode-thread"); + if (!NodeFS.existsSync(toolPath)) return null; + const result = NodeChildProcess.spawnSync(toolPath, ["list", "--json"], { + encoding: "utf8", + stdio: ["ignore", "pipe", "ignore"], + timeout: 10_000, + }); + if (result.error || result.status !== 0) return null; + try { + const parsed = JSON.parse(result.stdout); + return Array.isArray(parsed?.threads) ? parsed.threads : []; + } catch { + return null; + } +} + +function readLastCodeThread(home, threadId) { + const toolPath = NodePath.join(home, ".lastcode", "userdata", "bin", "lastcode-thread"); + if (!NodeFS.existsSync(toolPath)) return null; + const result = NodeChildProcess.spawnSync( + toolPath, + ["read", threadId, "--turn-limit", "1", "--json"], + { + encoding: "utf8", + stdio: ["ignore", "pipe", "ignore"], + timeout: 10_000, + }, + ); + if (result.error || result.status !== 0) return null; + try { + const parsed = JSON.parse(result.stdout); + return parsed?.kind === "read" ? parsed : null; + } catch { + return null; + } +} + +function readCheckpointThreads(home, config) { + const threads = readLastCodeThreads(home); + if (threads === null) return null; + const configuredId = + typeof config?.recoveryThreadId === "string" ? config.recoveryThreadId : undefined; + if (!configuredId || threads.some((thread) => thread.threadId === configuredId)) return threads; + const configuredThread = readLastCodeThread(home, configuredId); + return configuredThread === null ? threads : [...threads, configuredThread]; +} + +function repairPersistentThreadConfig(home) { + const { configPath, config } = readCheckpointSupervisorConfig(home); + if (!config) throw new Error("Checkpoint supervisor configuration is unavailable."); + const threads = readLastCodeThreads(home); + if (threads === null) throw new Error("LastCode thread state is unavailable."); + const persistentThreads = threads.filter((thread) => thread.persistent === true); + if (persistentThreads.length !== 1) { + throw new Error( + persistentThreads.length === 0 + ? "No persistent thread is designated. Mark one in LastCode, then retry." + : "More than one persistent thread was found; disable the extras before retrying.", + ); + } + const recoveryThreadId = persistentThreads[0].threadId; + const temporaryPath = `${configPath}.tmp`; + NodeFS.writeFileSync( + temporaryPath, + `${JSON.stringify({ ...config, recoveryThreadId }, null, 2)}\n`, + { encoding: "utf8", mode: 0o600 }, + ); + NodeFS.renameSync(temporaryPath, configPath); + console.log(`Repaired checkpoint recovery delivery to persistent thread ${recoveryThreadId}.`); } function parseNightly(tag) { @@ -878,6 +980,36 @@ function printDashboard(repoRoot, home, count, verbose) { for (const detail of carrySetShadowDetailLines(readRuns(home), verbose)) { console.log(style(detail.includes(" failed ") ? ansi.error : ansi.green, detail)); } + const { config: supervisorConfig } = readCheckpointSupervisorConfig(home); + const persistentStatus = persistentThreadRepairStatus( + supervisorConfig, + readCheckpointThreads(home, supervisorConfig), + ); + if (persistentStatus.kind === "stale" || persistentStatus.kind === "not-persistent") { + console.log( + style( + ansi.yellow, + persistentStatus.kind === "stale" + ? `Persistent thread safeguard: configured recovery thread ${persistentStatus.configuredId} no longer exists.` + : `Persistent thread safeguard: configured recovery thread ${persistentStatus.configuredId} is not marked persistent.`, + ), + ); + if (persistentStatus.replacementId) { + console.log( + style( + ansi.lavender, + `Repair to “${persistentStatus.replacementTitle}” (${persistentStatus.replacementId}): lastcode-checkpoints --repair-persistent-thread`, + ), + ); + } else { + console.log( + style( + ansi.lavender, + "Right-click the replacement thread in LastCode and choose “Mark as persistent thread”, then run lastcode-checkpoints --repair-persistent-thread.", + ), + ); + } + } console.log( style( freshness === "Up to date" @@ -899,7 +1031,9 @@ function printDashboard(repoRoot, home, count, verbose) { function main(argv) { const options = parseOptions(argv); if (options.help) { - console.log("Usage: lastcode-checkpoints [-n COUNT] [--verbose] [--repo PATH] [--install]"); + console.log( + "Usage: lastcode-checkpoints [-n COUNT] [--verbose] [--repo PATH] [--install] [--repair-persistent-thread]", + ); return; } const home = NodeOS.homedir(); @@ -908,6 +1042,10 @@ function main(argv) { installCommand(repoRoot, home); return; } + if (options.repairPersistentThread) { + repairPersistentThreadConfig(home); + return; + } printDashboard(repoRoot, home, options.count, options.verbose); } diff --git a/scripts/lastcode-checkpoints.test.mjs b/scripts/lastcode-checkpoints.test.mjs index dda8662b5f1b..f34d599b4e56 100644 --- a/scripts/lastcode-checkpoints.test.mjs +++ b/scripts/lastcode-checkpoints.test.mjs @@ -15,6 +15,7 @@ import { latestPublishedInstallableTag, parseRebaseRange, parseOptions, + persistentThreadRepairStatus, parseRemotePublicationState, parseRemoteUpstreamTags, parseTrailers, @@ -33,9 +34,38 @@ describe("LastCode checkpoint dashboard", () => { expect(parseOptions([]).count).toBe(8); expect(parseOptions(["-n", "12"]).count).toBe(12); expect(parseOptions(["--verbose"]).verbose).toBe(true); + expect(parseOptions(["--repair-persistent-thread"]).repairPersistentThread).toBe(true); expect(() => parseOptions(["-n", "0"])).toThrow("Invalid checkpoint count"); }); + it("detects stale and unprotected recovery thread configuration", () => { + const config = { schemaVersion: 1, recoveryThreadId: "thread-old" }; + expect(persistentThreadRepairStatus(config, [])).toEqual({ + kind: "stale", + configuredId: "thread-old", + }); + expect(persistentThreadRepairStatus(config, null)).toEqual({ + kind: "unavailable", + configuredId: "thread-old", + }); + expect( + persistentThreadRepairStatus(config, [ + { threadId: "thread-old", title: "Old", persistent: false }, + { threadId: "thread-new", title: "Maintenance", persistent: true }, + ]), + ).toEqual({ + kind: "not-persistent", + configuredId: "thread-old", + replacementId: "thread-new", + replacementTitle: "Maintenance", + }); + expect( + persistentThreadRepairStatus(config, [ + { threadId: "thread-old", title: "Maintenance", persistent: true }, + ]), + ).toEqual({ kind: "healthy", configuredId: "thread-old" }); + }); + it("parses checkpoint metadata trailers", () => { expect(parseTrailers("Title\n\nUpstream-Tag: v1-nightly.1\nDuration-Ms: 128000\n")).toEqual({ "Upstream-Tag": "v1-nightly.1",