diff --git a/apps/mobile/src/features/threads/ComposerMergeBackNotice.tsx b/apps/mobile/src/features/threads/ComposerMergeBackNotice.tsx new file mode 100644 index 000000000000..666591e60688 --- /dev/null +++ b/apps/mobile/src/features/threads/ComposerMergeBackNotice.tsx @@ -0,0 +1,31 @@ +import { pendingMergeBackNotice } from "@t3tools/client-runtime/state/thread-relationships"; +import { View } from "react-native"; + +import { AppText as Text } from "../../components/AppText"; +import { SymbolView } from "../../components/AppSymbol"; + +/** + * A merged-back fork changes what the next send carries. The notice leaves on + * its own once the next run consumes the transfer and the feed shows the + * context handoff divider. + */ +export function ComposerMergeBackNotice(props: { + readonly sourceThreadTitle: string | null; + readonly forkCount: number; + readonly waitsForIdle: boolean; +}) { + const notice = pendingMergeBackNotice(props); + return ( + + + + {notice.title} + {`. ${notice.description}`} + + + ); +} diff --git a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx index b6930deb5e86..1c41cd91f4fd 100644 --- a/apps/mobile/src/features/threads/ThreadDetailScreen.tsx +++ b/apps/mobile/src/features/threads/ThreadDetailScreen.tsx @@ -141,6 +141,7 @@ import { ComposerPopoverHost } from "./ComposerPopoverHost"; import { ThreadFeed, type ThreadFeedHistoryControls } from "./ThreadFeed"; import { useThreadTurnSubagents } from "./ThreadAgentsSheet"; import { ComposerQueuedEditBanner } from "./ComposerQueuedEdit"; +import { ComposerMergeBackNotice } from "./ComposerMergeBackNotice"; import { useThreadQueuedCount } from "./ThreadQueueControl"; import type { ThreadContentPresentation } from "./threadContentPresentation"; import { resolveThreadFeedSubmissionAnchor } from "./thread-feed-live-follow"; @@ -157,6 +158,12 @@ export interface ThreadDetailScreenProps { readonly environmentLabel: string | null; readonly feedbackSubmissions: ReadonlyArray; readonly onDismissFeedback: (id: MessageId) => void; + /** Set while a merged-back fork waits to be included in the next send. */ + readonly pendingMergeBack: { + readonly sourceThreadTitle: string | null; + readonly forkCount: number; + readonly waitsForIdle: boolean; + } | null; readonly selectedThreadFeed: ReadonlyArray; readonly activityRun: ThreadFeedLatestRun | null; readonly activeWorkStartedAt: string | null; @@ -1238,6 +1245,9 @@ export const ThreadDetailScreen = memo(function ThreadDetailScreen(props: Thread /> ) : null} + {props.pendingMergeBack !== null ? ( + + ) : null} resolvePendingMergeBack(selectedThreadDetail), + [selectedThreadDetail], + ); + // Primitives keep the memoized detail screen stable while the projection streams. + const mergeBackSourceThreadId = pendingMergeBack?.transfer.sourceThreadId ?? null; + const mergeBackForkCount = pendingMergeBack?.forkCount ?? 0; + const mergeBackWaitsForIdle = pendingMergeBack?.waitsForIdle ?? false; + const selectedEnvironmentId = selectedThread?.environmentId ?? null; + const pendingMergeBackSourceRef = useMemo( + () => + mergeBackSourceThreadId === null || selectedEnvironmentId === null + ? null + : scopeThreadRef(selectedEnvironmentId, mergeBackSourceThreadId), + [mergeBackSourceThreadId, selectedEnvironmentId], + ); + const pendingMergeBackSourceTitle = useThreadShell(pendingMergeBackSourceRef)?.title ?? null; + const pendingMergeBackNotice = useMemo( + () => + mergeBackSourceThreadId === null + ? null + : { + sourceThreadTitle: pendingMergeBackSourceTitle, + forkCount: mergeBackForkCount, + waitsForIdle: mergeBackWaitsForIdle, + }, + [ + mergeBackForkCount, + mergeBackSourceThreadId, + mergeBackWaitsForIdle, + pendingMergeBackSourceTitle, + ], + ); const handleMergeBack = useCallback(async () => { if ( mergeBackBusyRef.current || @@ -938,6 +976,7 @@ function ThreadRouteContent( environmentLabel={selectedEnvironmentConnection?.environmentLabel ?? null} feedbackSubmissions={composer.feedbackSubmissions} onDismissFeedback={composer.dismissFeedback} + pendingMergeBack={pendingMergeBackNotice} selectedThreadFeed={composer.selectedThreadFeed} activityRun={composer.selectedThreadActivityRun} activeWorkStartedAt={ diff --git a/apps/web/src/components/ChatView.tsx b/apps/web/src/components/ChatView.tsx index 69a204ccd8a3..3b4f11c9eeb9 100644 --- a/apps/web/src/components/ChatView.tsx +++ b/apps/web/src/components/ChatView.tsx @@ -37,6 +37,7 @@ import { isUsageLimitsCommand, } from "@t3tools/shared/usageLimits"; import { feedbackBannerItem } from "./chat/ComposerFeedback"; +import { mergeBackBannerItem } from "./chat/ComposerMergeBackNotice"; import { usageLimitsBannerItem } from "./chat/ComposerUsageLimits"; import { getTerminalLabel } from "@t3tools/shared/terminalLabels"; import * as Schema from "effect/Schema"; @@ -107,6 +108,7 @@ import { presentProviderGoal, } from "@t3tools/client-runtime/state/thread-execution"; import { threadSupportsProviderHandoff } from "@t3tools/client-runtime/state/thread-workflows"; +import { resolvePendingMergeBack } from "@t3tools/client-runtime/state/thread-relationships"; import { codexFeedbackMessage, parseCodexFeedbackCommand, @@ -7603,6 +7605,35 @@ export default function ChatView(props: ChatViewProps) { isStoppingBackgroundWork, onOpenRelatedThread, ]); + const pendingMergeBack = useMemo( + () => resolvePendingMergeBack(serverProjection), + [serverProjection], + ); + // Primitives keep the banner stable while the projection streams. + const mergeBackTransferId = pendingMergeBack?.transfer.id ?? null; + const mergeBackSourceThreadId = pendingMergeBack?.transfer.sourceThreadId ?? null; + const mergeBackForkCount = pendingMergeBack?.forkCount ?? 0; + const mergeBackWaitsForIdle = pendingMergeBack?.waitsForIdle ?? false; + const mergeBackSourceRef = useMemo( + () => + mergeBackSourceThreadId === null + ? null + : scopeThreadRef(environmentId, mergeBackSourceThreadId), + [environmentId, mergeBackSourceThreadId], + ); + const mergeBackSourceTitle = useThreadShell(mergeBackSourceRef)?.title ?? null; + const mergeBackBanner = useMemo( + () => + mergeBackTransferId === null + ? null + : mergeBackBannerItem({ + transferId: mergeBackTransferId, + sourceThreadTitle: mergeBackSourceTitle, + forkCount: mergeBackForkCount, + waitsForIdle: mergeBackWaitsForIdle, + }), + [mergeBackForkCount, mergeBackSourceTitle, mergeBackTransferId, mergeBackWaitsForIdle], + ); // Settled, snoozed, and woke are thread state, not composer actions: each // gets one quiet line after the last message instead of a banner. A woken // thread announces itself here, not just in the sidebar pill. Dismissing @@ -7780,12 +7811,14 @@ export default function ChatView(props: ChatViewProps) { ); // The user asked for this one, so it leads the notice tier instead of trailing it. const usageLimitsItems = usageLimitsBanner === null ? [] : [usageLimitsBanner]; + const mergeBackItems = mergeBackBanner === null ? [] : [mergeBackBanner]; const projectCloneItems = projectCloneBannerItem === null ? [] : [projectCloneBannerItem]; if (!localCheckoutBranchMismatch || !showBranchMismatchBanner || !activeBranchMismatchKey) { return [ ...feedbackBannerItems, ...limitRecoveryItems, ...usageLimitsItems, + ...mergeBackItems, ...projectCloneItems, ...systemComposerBannerItems, ...backgroundWorkItems, @@ -7795,6 +7828,7 @@ export default function ChatView(props: ChatViewProps) { ...feedbackBannerItems, ...limitRecoveryItems, ...usageLimitsItems, + ...mergeBackItems, ...projectCloneItems, ...systemComposerBannerItems, ...backgroundWorkItems, @@ -7849,6 +7883,7 @@ export default function ChatView(props: ChatViewProps) { backgroundWorkBannerItem, goalBannerItem, localCheckoutBranchMismatch, + mergeBackBanner, projectCloneBannerItem, showBranchMismatchBanner, systemComposerBannerItems, diff --git a/apps/web/src/components/chat/ComposerMergeBackNotice.tsx b/apps/web/src/components/chat/ComposerMergeBackNotice.tsx new file mode 100644 index 000000000000..5ae3f7917f2c --- /dev/null +++ b/apps/web/src/components/chat/ComposerMergeBackNotice.tsx @@ -0,0 +1,25 @@ +import { pendingMergeBackNotice } from "@t3tools/client-runtime/state/thread-relationships"; +import type { ContextTransferId } from "@t3tools/contracts"; + +import { PullRequestGlyph } from "../pullRequest/pullRequestIcons"; +import type { ComposerBannerStackItem } from "./ComposerBannerStack"; + +/** + * A merged-back fork changes what the next send carries, so it gets a composer + * notice. It has no dismiss: the next run consumes the transfer and the + * timeline's context handoff divider takes over. + */ +export function mergeBackBannerItem(input: { + readonly transferId: ContextTransferId; + readonly sourceThreadTitle: string | null; + readonly forkCount: number; + readonly waitsForIdle: boolean; +}): ComposerBannerStackItem { + const { blocked, ...notice } = pendingMergeBackNotice(input); + return { + id: `merge-back:${input.transferId}`, + variant: blocked ? "warning" : "info", + icon: , + ...notice, + }; +} diff --git a/packages/client-runtime/src/state/threadRelationships.test.ts b/packages/client-runtime/src/state/threadRelationships.test.ts index d96fba5aab22..11d8ed192ad2 100644 --- a/packages/client-runtime/src/state/threadRelationships.test.ts +++ b/packages/client-runtime/src/state/threadRelationships.test.ts @@ -1,5 +1,12 @@ import { describe, expect, it } from "vite-plus/test"; -import { ThreadId } from "@t3tools/contracts"; +import { + ContextTransferId, + ProjectId, + ProviderInstanceId, + ThreadId, + type OrchestrationV2ContextTransfer, + type OrchestrationV2ThreadProjection, +} from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; import { @@ -8,6 +15,9 @@ import { orderWebThreadLineageRows, relatedThreadIds, resolveMergeBackTargetThreadId, + pendingMergeBackNotice, + resolvePendingMergeBack, + resolvePendingMergeBackTransfer, walkThreadRelationships, threadRelationshipRowStatus, } from "./threadRelationships.ts"; @@ -503,3 +513,177 @@ describe("web thread lineage ordering", () => { ).toEqual([first, second, third]); }); }); + +describe("pending merge-back", () => { + const target = ThreadId.make("thread-target"); + const fork = ThreadId.make("thread-fork"); + const now = DateTime.makeUnsafe("2026-06-20T00:00:00.000Z"); + const thread = { + id: target, + projectId: ProjectId.make("project"), + title: "Source thread", + providerInstanceId: ProviderInstanceId.make("claudeAgent"), + modelSelection: { instanceId: ProviderInstanceId.make("claudeAgent"), model: "sonnet" }, + runtimeMode: "full-access", + interactionMode: "default", + branch: null, + worktreePath: null, + activeProviderThreadId: null, + lineage: { rootThreadId: target, parentThreadId: null, relationshipToParent: null }, + forkedFrom: null, + createdBy: "user", + creationSource: "web", + createdAt: now, + updatedAt: now, + archivedAt: null, + settledOverride: null, + settledAt: null, + lastVisitedAt: null, + deletedAt: null, + } satisfies OrchestrationV2ThreadProjection["thread"]; + + function transfer(id: string, overrides: Partial = {}) { + return { + id: ContextTransferId.make(id), + type: "merge_back", + status: "pending", + sourceThreadId: fork, + targetThreadId: target, + sourcePoint: { threadId: fork }, + basePoint: null, + sourceProviderInstanceId: null, + targetProviderInstanceId: null, + targetRunId: null, + resolution: null, + createdBy: "user", + error: null, + createdAt: now, + updatedAt: now, + consumedAt: null, + ...overrides, + } satisfies OrchestrationV2ContextTransfer; + } + + function resolve(contextTransfers: ReadonlyArray) { + return resolvePendingMergeBackTransfer({ thread, contextTransfers }); + } + + it("returns null for a missing projection or no transfers", () => { + expect(resolvePendingMergeBackTransfer(null)).toBeNull(); + expect(resolve([])).toBeNull(); + }); + + it("selects by updatedAt rather than creation or array order without copying or sorting", () => { + const newer = Object.freeze( + transfer("newer", { + updatedAt: DateTime.makeUnsafe("2026-06-20T00:02:00.000Z"), + }), + ); + const older = Object.freeze( + transfer("older", { + createdAt: DateTime.makeUnsafe("2026-06-20T00:03:00.000Z"), + updatedAt: DateTime.makeUnsafe("2026-06-20T00:01:00.000Z"), + }), + ); + expect(resolve(Object.freeze([newer, older]))).toBe(newer); + expect(resolve(Object.freeze([older, newer]))).toBe(newer); + }); + + it("matches the server's last-entry-wins tie-break for equal updatedAt", () => { + const first = transfer("first"); + const second = transfer("second"); + expect(resolve([first, second])).toBe(second); + expect(resolve([second, first])).toBe(first); + }); + + it.each(["resolved_native", "resolved_portable", "failed", "consumed", "superseded"] as const)( + "ignores a newer %s transfer", + (status) => { + const pending = transfer("pending"); + const settled = transfer("settled", { + status, + updatedAt: DateTime.makeUnsafe("2026-06-20T00:05:00.000Z"), + }); + expect(resolve([pending, settled])).toBe(pending); + expect(resolve([settled])).toBeNull(); + }, + ); + + it.each(["fork", "provider_handoff", "subagent_spawn", "subagent_result"] as const)( + "ignores a pending %s transfer", + (type) => { + expect(resolve([transfer("other", { type })])).toBeNull(); + }, + ); + + it("ignores outgoing transfers and transfers for a different target", () => { + expect( + resolve([ + transfer("outgoing", { sourceThreadId: target, targetThreadId: fork }), + transfer("elsewhere", { targetThreadId: ThreadId.make("elsewhere") }), + ]), + ).toBeNull(); + }); + + it("replaces a superseded merge-back and clears when the replacement is consumed", () => { + const first = transfer("first"); + expect(resolve([first])).toBe(first); + const superseded = { ...first, status: "superseded" as const }; + const replacement = transfer("replacement"); + expect(resolve([superseded, replacement])).toBe(replacement); + expect(resolve([superseded, { ...replacement, status: "consumed" }])).toBeNull(); + }); + + function resolveState( + contextTransfers: ReadonlyArray, + runStatuses: ReadonlyArray = [], + ) { + return resolvePendingMergeBack({ + thread, + contextTransfers, + runs: runStatuses.map((status) => ({ status })) as never, + }); + } + + it("expects the next send to carry the transfer on an idle thread", () => { + const pending = transfer("pending"); + expect(resolveState([pending], ["completed", "interrupted"])).toEqual({ + transfer: pending, + forkCount: 1, + waitsForIdle: false, + }); + expect(resolveState([])).toBeNull(); + expect(resolvePendingMergeBack(null)).toBeNull(); + }); + + it.each(["preparing", "starting", "running", "waiting"])( + "waits for idle while a run is %s, as the server rejects or steers that send", + (status) => { + expect(resolveState([transfer("pending")], ["completed", status])?.waitsForIdle).toBe(true); + }, + ); + + it("counts distinct pending forks, which the server rejects together", () => { + const otherFork = ThreadId.make("thread-other-fork"); + const later = DateTime.makeUnsafe("2026-06-20T00:01:00.000Z"); + const state = resolveState([ + transfer("fork-older", { status: "superseded" }), + transfer("fork-newer"), + transfer("other-fork", { sourceThreadId: otherFork, updatedAt: later }), + transfer("third-fork-done", { sourceThreadId: ThreadId.make("done"), status: "consumed" }), + ]); + expect(state?.forkCount).toBe(2); + expect(state?.transfer.id).toBe("other-fork"); + expect(resolveState([transfer("a"), transfer("b", { updatedAt: later })])?.forkCount).toBe(1); + }); + + it("only marks the notice as blocking when more than one fork is pending", () => { + const notice = (forkCount: number, waitsForIdle: boolean) => + pendingMergeBackNotice({ sourceThreadTitle: "Fork", forkCount, waitsForIdle }); + expect(notice(1, false).blocked).toBe(false); + expect(notice(1, true).blocked).toBe(false); + expect(notice(1, true).description).not.toBe(notice(1, false).description); + expect(notice(2, false).blocked).toBe(true); + expect(notice(2, true).blocked).toBe(true); + }); +}); diff --git a/packages/client-runtime/src/state/threadRelationships.ts b/packages/client-runtime/src/state/threadRelationships.ts index 4ef792a6ada7..c9f561113a5f 100644 --- a/packages/client-runtime/src/state/threadRelationships.ts +++ b/packages/client-runtime/src/state/threadRelationships.ts @@ -1,10 +1,13 @@ import type { + OrchestrationV2ContextTransfer, OrchestrationV2ThreadProjection, OrchestrationV2ThreadShell, ThreadId, } from "@t3tools/contracts"; import * as DateTime from "effect/DateTime"; +import { resolveActiveThreadRun } from "./threadWorkflows.ts"; + export type ThreadRelationshipKind = "parent" | "fork" | "subagent" | "transfer"; export interface ThreadRelationshipNode { @@ -41,6 +44,86 @@ export function resolveMergeBackTargetThreadId( : projection.thread.lineage.parentThreadId; } +/** + * The merge-back the next send on this thread will carry, or null. Mirrors the + * server, which consumes only the newest pending merge-back targeting the thread. + */ +export function resolvePendingMergeBackTransfer( + projection: Pick | null, +): OrchestrationV2ContextTransfer | null { + if (projection === null) return null; + let latest: OrchestrationV2ContextTransfer | null = null; + for (const transfer of projection.contextTransfers) { + if ( + transfer.type !== "merge_back" || + transfer.status !== "pending" || + transfer.targetThreadId !== projection.thread.id + ) { + continue; + } + if ( + latest === null || + DateTime.toEpochMillis(transfer.updatedAt) >= DateTime.toEpochMillis(latest.updatedAt) + ) { + latest = transfer; + } + } + return latest; +} + +export interface PendingMergeBack { + readonly transfer: OrchestrationV2ContextTransfer; + /** Distinct forks with a pending merge-back here. The server rejects sends while this exceeds 1. */ + readonly forkCount: number; + /** A run is active. The server only consumes the transfer on a send made once the thread is idle. */ + readonly waitsForIdle: boolean; +} + +/** The pending merge-back on this thread, with what keeps the next send from carrying it. */ +export function resolvePendingMergeBack( + projection: Pick | null, +): PendingMergeBack | null { + const transfer = resolvePendingMergeBackTransfer(projection); + if (projection === null || transfer === null) return null; + const forks = new Set(); + for (const candidate of projection.contextTransfers) { + if ( + candidate.type === "merge_back" && + candidate.status === "pending" && + candidate.targetThreadId === projection.thread.id + ) { + forks.add(candidate.sourceThreadId); + } + } + return { + transfer, + forkCount: forks.size, + waitsForIdle: resolveActiveThreadRun(projection) !== null, + }; +} + +/** Copy for the composer notice shown while a merge-back waits for a send. */ +export function pendingMergeBackNotice(input: { + readonly sourceThreadTitle: string | null; + readonly forkCount: number; + readonly waitsForIdle: boolean; +}) { + if (input.forkCount > 1) { + return { + blocked: true, + title: `Merged back from ${input.forkCount} forks`, + description: "Sending will fail while more than one merged fork is pending", + }; + } + return { + blocked: false, + title: `Merged back from ${input.sourceThreadTitle ?? "a fork"}`, + description: input.waitsForIdle + ? "Its context will be included in the next message you send once this thread is idle" + : "Its context will be included in your next message", + }; +} + function edgeKey(edge: ThreadRelationshipEdge): string { return `${edge.sourceThreadId}\u001f${edge.targetThreadId}\u001f${edge.kind}`; }