fix(orchestration): stabilize Codex turn mapping and settlement

- Preserve Codex turn identity while suppressing duplicate diff notifications - Optimize settlement projections and isolate thread visit handling - Add concurrency and regression coverage across server and mobile

Julius Marminge committed 2 months ago

Commit 6502e221 parent 25d2531167 files · +189949 −140Browse files

Showing the changes to apps/web/src (6 of 167 files). Show all

Unified · Split

6 files

apps/web/src/components/CommandPalette.tsx +9 −1
apps/web/src/components/ThreadStatusIndicators.subscriptions.test.tsx +180 −0
apps/web/src/components/ThreadStatusIndicators.test.ts +19 −0
apps/web/src/components/ThreadStatusIndicators.tsx +59 −17
apps/web/src/connection/desktopLocal.test.ts +62 −1
apps/web/src/connection/desktopLocal.ts +19 −2

apps/web/src/components/CommandPalette.tsx+9−1

         projectTitleById,
         sortOrder: clientSettings.sidebarThreadSortOrder,
         icon: <MessageSquareIcon className={ITEM_ICON_CLASS} />,
-        renderLeadingContent: (thread) => <ThreadRowLeadingStatus thread={thread} />,
+        renderLeadingContent: (thread) => (
+          <ThreadRowLeadingStatus
+            thread={thread}
+            snapshot={changeRequestSnapshotByKey.get(
+              scopedThreadKey(scopeThreadRef(thread.environmentId, thread.id)),
+            )}
+          />
+        ),
         renderTrailingContent: (thread) => <ThreadRowTrailingStatus thread={thread} />,
         renderDescription: (thread, { projectTitle }) => {
           const modelInstanceId =
       }),
     [
       activeThreadId,
+      changeRequestSnapshotByKey,
       clientSettings.sidebarThreadSortOrder,
       navigate,
       projectCwdById,

apps/web/src/components/ThreadStatusIndicators.subscriptions.test.tsx+180−0

@@ -0,0 +1,180 @@
	1	import { ProjectId, ThreadId, type VcsStatusResult } from "@t3tools/contracts";
	2	import { act, type ComponentProps, type ReactNode } from "react";
	3	import { create, type ReactTestRenderer } from "react-test-renderer";
	4	import { afterEach, expect, it, vi } from "vite-plus/test";
	5
	6	import { AppAtomRegistryProvider } from "../rpc/atomRegistry";
	7	import { makeThreadFixture } from "../test-fixtures";
	8
	9	const state = vi.hoisted(() => ({ queries: [] as string[] }));
	10
	11	vi.mock("../state/entities", () => ({ useProject: () => null }));
	12	vi.mock("../uiStateStore", () => ({ useUiStateStore: () => undefined }));
	13	vi.mock("../state/vcs", async () => {
	14	const { AsyncResult, Atom } = await import("effect/unstable/reactivity");
	15	const statuses = Atom.family((cwd: string) =>
	16	Atom.make(() => {
	17	state.queries.push(`vcs:${cwd}`);
	18	return AsyncResult.success({
	19	isRepo: true,
	20	hasPrimaryRemote: true,
	21	isDefaultRef: false,
	22	refName: "feature/test",
	23	hasWorkingTreeChanges: false,
	24	workingTree: { files: [], insertions: 0, deletions: 0 },
	25	hasUpstream: true,
	26	aheadCount: 0,
	27	behindCount: 0,
	28	pr: {
	29	number: 42,
	30	title: "Observed worktree PR",
	31	url: "https://github.com/example/repo/pull/42",
	32	baseRef: "main",
	33	headRef: "feature/test",
	34	state: "open",
	35	},
	36	} satisfies VcsStatusResult);
	37	}),
	38	);
	39	return {
	40	vcsEnvironment: { status: ({ input }: { input: { cwd: string } }) => statuses(input.cwd) },
	41	};
	42	});
	43	vi.mock("../state/pullRequests", async (importOriginal) => {
	44	const original = await importOriginal<typeof import("../state/pullRequests")>();
	45	const { AsyncResult, Atom } = await import("effect/unstable/reactivity");
	46	const { ProjectId } = await import("@t3tools/contracts");
	47	const summaries = Atom.family((number: number) =>
	48	Atom.make(() => {
	49	state.queries.push(`pr:${number}`);
	50	return AsyncResult.success({
	51	provider: "github" as const,
	52	projectId: ProjectId.make("project-test"),
	53	repository: "example/repo",
	54	number,
	55	title: "Observed linked PR",
	56	url: `https://github.com/example/repo/pull/${number}`,
	57	state: "open" as const,
	58	headBranch: "feature/test",
	59	baseBranch: "main",
	60	updatedAt: "2026-09-04T00:00:00.000Z",
	61	});
	62	}),
	63	);
	64	return {
	65	...original,
	66	linkedPullRequestDetailAtom: ({ input }: { input: { number: number } }) =>
	67	summaries(input.number),
	68	};
	69	});
	70	vi.mock("./ui/tooltip", async () => {
	71	const { cloneElement, isValidElement } = await import("react");
	72	return {
	73	Tooltip: ({ children }: { children: ReactNode }) => <>{children}</>,
	74	TooltipTrigger({
	75	render,
	76	children,
	77	}: ComponentProps<typeof import("./ui/tooltip").TooltipTrigger>) {
	78	return isValidElement(render) ? cloneElement(render, undefined, children) : children;
	79	},
	80	TooltipPopup: () => null,
	81	};
	82	});
	83
	84	import { ThreadRowLeadingStatus } from "./ThreadStatusIndicators";
	85
	86	afterEach(() => {
	87	vi.unstubAllGlobals();
	88	});
	89
	90	it("leases only visible palette results while retaining observed and cached PR badges", async () => {
	91	const observers: Array<{ setVisible: (visible: boolean) => void }> = [];
	92	const scrollRoot = {};
	93	vi.stubGlobal("IS_REACT_ACT_ENVIRONMENT", true);
	94	vi.stubGlobal(
	95	"IntersectionObserver",
	96	class {
	97	constructor(callback: (entries: ReadonlyArray<{ isIntersecting: boolean }>) => void) {
	98	observers.push({ setVisible: (visible) => callback([{ isIntersecting: visible }]) });
	99	}
	100	observe() {}
	101	disconnect() {}
	102	},
	103	);
	104	const threads = Array.from({ length: 128 }, (_, index) =>
	105	makeThreadFixture({
	106	id: ThreadId.make(`palette-thread-${index}`),
	107	branch: "feature/test",
	108	worktreePath: `/test/worktree-${index}`,
	109	linkedPullRequest:
	110	index % 2 === 0
	111	? null
	112	: {
	113	projectId: ProjectId.make("project-test"),
	114	repository: "example/repo",
	115	number: index,
	116	url: `https://github.com/example/repo/pull/${index}`,
	117	},
	118	}),
	119	);
	120	let renderer: ReactTestRenderer \| undefined;
	121	state.queries.length = 0;
	122	try {
	123	await act(async () => {
	124	renderer = create(
	125	<AppAtomRegistryProvider>
	126	{threads.map((thread, index) => (
	127	<span key={thread.id}>
	128	<ThreadRowLeadingStatus
	129	thread={thread}
	130	snapshot={
	131	index === 126
	132	? {
	133	branch: "feature/test",
	134	sourceControlProvider: undefined,
	135	pr: {
	136	number: 126,
	137	title: "Cached badge",
	138	url: "https://github.com/example/repo/pull/126",
	139	baseRef: "main",
	140	headRef: "feature/test",
	141	state: "open",
	142	},
	143	}
	144	: undefined
	145	}
	146	/>
	147	{thread.title}
	148	</span>
	149	))}
	150	</AppAtomRegistryProvider>,
	151	{ createNodeMock: () => ({ parentElement: { closest: () => scrollRoot } }) },
	152	);
	153	});
	154	const mounted = renderer!;
	155	const badgeCount = () =>
	156	mounted.root.findAll(
	157	(node) => node.type === "span" && String(node.props["aria-label"] ?? "").startsWith("PR #"),
	158	).length;
	159	expect(observers).toHaveLength(128);
	160	expect(state.queries).toEqual([]);
	161	expect(badgeCount()).toBe(1);
	162
	163	await act(async () => {
	164	for (const observer of observers.slice(0, 8)) observer.setVisible(true);
	165	});
	166	expect(state.queries).toHaveLength(8);
	167	expect(state.queries.filter((key) => key.startsWith("vcs:"))).toHaveLength(4);
	168	expect(state.queries.filter((key) => key.startsWith("pr:"))).toHaveLength(4);
	169	expect(badgeCount()).toBe(9);
	170
	171	await act(async () => {
	172	for (const observer of observers.slice(0, 8)) observer.setVisible(false);
	173	for (const observer of observers.slice(8, 16)) observer.setVisible(true);
	174	});
	175	expect(state.queries).toHaveLength(16);
	176	expect(badgeCount()).toBe(17);
	177	} finally {
	178	await act(async () => renderer?.unmount());
	179	}
	180	});

apps/web/src/components/ThreadStatusIndicators.test.ts+19−0

@@ -554,6 +554,25 @@ describe("resolveDisplayedThreadPr + nextThreadChangeRequestSnapshot", () => { · 553 unmodified lines
554	).toEqual(mergedPr);	554	).toEqual(mergedPr);
555	});	555	});
556		556
		557	it("shows the last matching PR snapshot while its live status lease is released", () => {
		558	const openPr = { ...mergedPr, state: "open" as const };
		559	const input = {
		560	threadBranch: featureBranch,
		561	gitStatus: null,
		562	snapshot: snapshotFor(featureBranch, openPr, provider),
		563	retainTerminalOnBranchMismatch: false,
		564	};
		565	expect(resolveDisplayedThreadPr(input)).toEqual(openPr);
		566	expect(resolveDisplayedThreadPrProvider(input)).toEqual(provider);
		567	expect(resolveDisplayedThreadPr({ ...input, threadBranch: "feature/other" })).toBeNull();
		568	expect(
		569	resolveDisplayedThreadPr({
		570	...input,
		571	snapshot: { ...input.snapshot, linkedPullRequest },
		572	}),
		573	).toBeNull();
		574	});
		575
557	it("retains a merged PR after a main checkout", () => {	576	it("retains a merged PR after a main checkout", () => {
558	const matchingStatus = status({	577	const matchingStatus = status({
559	refName: featureBranch,	578	refName: featureBranch,

apps/web/src/components/ThreadStatusIndicators.tsx+59−17

 } from "@t3tools/contracts";
 import { Atom } from "effect/unstable/reactivity";
 import { FolderGit2Icon, TerminalIcon } from "lucide-react";
-import { useMemo } from "react";
+import { useCallback, useMemo } from "react";
 import { appAtomRegistry } from "../rpc/atomRegistry";
 import { useEnvironment, usePrimaryEnvironmentId } from "../state/environments";
 import { EnvironmentMachineIcon } from "./EnvironmentMachineIcon";
   resolveThreadLastVisitedAt,
   resolveThreadStatusPill,
   type ThreadStatusPill,
+  useRetainedValue,
+  useSidebarRowSubscriptionLease,
 } from "./Sidebar.logic";
 import { resolvePullRequestState } from "./pullRequest/pullRequestPresentation";
 import type { SidebarThreadSummary } from "../types";
 export function useLinkedThreadPullRequest(
   environmentId: EnvironmentId | null,
   linkedPullRequest: ThreadLinkedPullRequest | null | undefined,
+  enabled = true,
 ): LinkedThreadPullRequestStatus | null {
   const queried = useEnvironmentQuery(
-    environmentId === null || linkedPullRequest == null
+    !enabled || environmentId === null || linkedPullRequest == null
       ? null
       : linkedPullRequestDetailAtom({
           environmentId,
  * checkouts only, a cached merged/closed PR for the thread. Local thread
  * metadata follows the shared checkout, so the cached branch intentionally
  * survives that metadata changing to the newly checked-out branch. Open PRs
- * are never retained — their state can still change.
+ * are retained only while live status is absent and the branch still matches.
  */
 export function resolveDisplayedThreadPr(input: {
   threadBranch: string | null;
     return gitStatus.pr;
   }
+  if (
+    gitStatus === null &&
+    threadBranch !== null &&
+    snapshot?.branch === threadBranch &&
+    snapshot.linkedPullRequest === undefined
+  ) {
+    return snapshot.pr;
+  }
   if (
     threadBranch !== null &&
     retainTerminalOnBranchMismatch &&
     return gitStatus.sourceControlProvider;
   }
+  if (
+    gitStatus === null &&
+    threadBranch !== null &&
+    snapshot?.branch === threadBranch &&
+    snapshot.linkedPullRequest === undefined
+  ) {
+    return snapshot.sourceControlProvider;
+  }
   if (
     threadBranch !== null &&
     retainTerminalOnBranchMismatch &&
  * like the command palette. Shows the change request state icon (if present) and the
  * thread status dot, matching the sidebar's leading indicators.
  */
-export function ThreadRowLeadingStatus({ thread }: { thread: SidebarThreadSummary }) {
+export function ThreadRowLeadingStatus({
+  thread,
+  snapshot,
+}: {
+  thread: SidebarThreadSummary;
+  snapshot?: ThreadChangeRequestSnapshot | undefined;
+}) {
+  const { leaseLiveStatus, rowRef } = useSidebarRowSubscriptionLease(false);
+  // Observe the containing title even when this thread has no badge yet.
+  const statusRef = useCallback(
+    (node: HTMLSpanElement | null) => rowRef(node?.parentElement ?? null),
+    [rowRef],
+  );
   const threadRef = scopeThreadRef(thread.environmentId, thread.id);
   const localLastVisitedAt = useUiStateStore(
     (state) => state.threadLastVisitedAtById[scopedThreadKey(threadRef)],
   const linkedPullRequest = useLinkedThreadPullRequest(
     thread.environmentId,
     thread.linkedPullRequest,
+    leaseLiveStatus,
   );
   const gitStatus = useEnvironmentQuery(
-    thread.linkedPullRequest == null &&
+    leaseLiveStatus &&
+      thread.linkedPullRequest == null &&
       (thread.branch != null || thread.worktreePath !== null) &&
       gitCwd !== null
       ? vcsEnvironment.status({
         })
       : null,
   );
-  const pr =
+  const visibleGitStatus = useRetainedValue(
-    thread.linkedPullRequest == null
+    JSON.stringify([thread.environmentId, gitCwd]),
-      ? resolveThreadPr({ threadBranch: thread.branch, gitStatus: gitStatus.data })
+    gitStatus.data,
-      : (linkedPullRequest?.pr ?? null);
-  const prStatus = prStatusIndicator(
-    pr,
-    linkedPullRequest?.sourceControlProvider ?? gitStatus.data?.sourceControlProvider,
   );
+  const displayedPrInput = {
+    threadBranch: thread.branch,
+    gitStatus: visibleGitStatus,
+    snapshot,
+    retainTerminalOnBranchMismatch: thread.worktreePath === null,
+    linkedPullRequest: thread.linkedPullRequest,
+    linkedPullRequestStatus: linkedPullRequest,
+  };
+  const pr = resolveDisplayedThreadPr(displayedPrInput);
+  const prStatus = prStatusIndicator(pr, resolveDisplayedThreadPrProvider(displayedPrInput));
   const threadStatus = resolveThreadStatusPill({
     thread: {
       ...thread,
     },
   });
-  if (!prStatus && !threadStatus) {
-    return null;
-  }
   return (
-    <span className="inline-flex shrink-0 items-center gap-1.5">
+    <span
+      ref={statusRef}
+      className={
+        prStatus || threadStatus ? "inline-flex shrink-0 items-center gap-1.5" : "contents"
+      }
+    >
       {prStatus && pr ? (
         <Tooltip>
           <TooltipTrigger

apps/web/src/connection/desktopLocal.test.ts+62−1

@@ -2,7 +2,11 @@ import { · 1 unmodified line
2	BearerConnectionTarget,	2	BearerConnectionTarget,
3	PrimaryConnectionTarget,	3	PrimaryConnectionTarget,
4	} from "@t3tools/client-runtime/connection";	4	} from "@t3tools/client-runtime/connection";
5	import { EnvironmentId, PRIMARY_LOCAL_ENVIRONMENT_ID } from "@t3tools/contracts";	5	import {
		6	type DesktopEnvironmentBootstrap,
		7	EnvironmentId,
		8	PRIMARY_LOCAL_ENVIRONMENT_ID,
		9	} from "@t3tools/contracts";
6	import { describe, expect, it } from "vite-plus/test";	10	import { describe, expect, it } from "vite-plus/test";
7		11
8	import {	12	import {
@@ -38,6 +42,63 @@ describe("desktop local connection identity", () => { · 29 unmodified lines
38	});	42	});
39		43
40	describe("desktop local topology reads", () => {	44	describe("desktop local topology reads", () => {
		45	it("reuses snapshots when polling returns fresh objects for the same topology", () => {
		46	const secondary: DesktopEnvironmentBootstrap = {
		47	id: "wsl:default",
		48	label: "WSL",
		49	runningDistro: "Ubuntu",
		50	httpBaseUrl: "http://127.0.0.1:4000",
		51	wsBaseUrl: "ws://127.0.0.1:4000",
		52	bootstrapToken: "bootstrap-1",
		53	};
		54	let entries = [secondary];
		55	const reader = createDesktopSecondaryBootstrapsReader(() => ({
		56	getLocalEnvironmentBootstraps: () => entries.map((entry) => ({ ...entry })),
		57	}));
		58
		59	const connected = reader.readSnapshot();
		60	expect(reader.readSnapshot()).toBe(connected);
		61	expect(reader.readResult()).toEqual({ _tag: "Success", bootstraps: connected });
		62	expect(reader.readSnapshot()).toBe(connected);
		63
		64	entries = [];
		65	const empty = reader.readSnapshot();
		66	expect(empty).toEqual([]);
		67	expect(reader.readSnapshot()).toBe(empty);
		68	});
		69
		70	it.each<Partial<DesktopEnvironmentBootstrap>>([
		71	{ id: "wsl:Debian" },
		72	{ label: "Renamed backend" },
		73	{ runningDistro: "Debian" },
		74	{ httpBaseUrl: "http://127.0.0.1:4001" },
		75	{ wsBaseUrl: "ws://127.0.0.1:4001" },
		76	{ bootstrapToken: "bootstrap-2" },
		77	])("publishes bootstrap changes: %j", (change) => {
		78	let secondary: DesktopEnvironmentBootstrap = {
		79	id: "wsl:default",
		80	label: "WSL",
		81	runningDistro: "Ubuntu",
		82	httpBaseUrl: "http://127.0.0.1:4000",
		83	wsBaseUrl: "ws://127.0.0.1:4000",
		84	bootstrapToken: "bootstrap-1",
		85	};
		86	const reader = createDesktopSecondaryBootstrapsReader(() => ({
		87	getLocalEnvironmentBootstraps: () => [{ ...secondary }],
		88	}));
		89	const before = reader.readSnapshot();
		90	secondary = { ...secondary, ...change };
		91	const after = reader.readSnapshot();
		92	expect(after).not.toBe(before);
		93	expect(after).toEqual([secondary]);
		94	expect(reader.readSnapshot()).toBe(after);
		95	});
		96
		97	it("keeps the empty snapshot stable without a desktop bridge", () => {
		98	const reader = createDesktopSecondaryBootstrapsReader(() => undefined);
		99	expect(reader.readSnapshot()).toBe(reader.readSnapshot());
		100	});
		101
41	it("distinguishes a successful empty topology from a read failure", () => {	102	it("distinguishes a successful empty topology from a read failure", () => {
42	let readBootstraps = () => [];	103	let readBootstraps = () => [];
43	const reader = createDesktopSecondaryBootstrapsReader(() => ({	104	const reader = createDesktopSecondaryBootstrapsReader(() => ({

apps/web/src/connection/desktopLocal.ts+19−2

@@ -67,13 +67,30 @@ export function createDesktopSecondaryBootstrapsReader( · 66 unmodified lines
67	const readResult = (): DesktopSecondaryBootstrapsRead => {	67	const readResult = (): DesktopSecondaryBootstrapsRead => {
68	const bridge = resolveBridge();	68	const bridge = resolveBridge();
69	if (bridge === undefined) {	69	if (bridge === undefined) {
70	snapshot = [];	70	if (snapshot.length > 0) snapshot = [];
71	return { _tag: "Success", bootstraps: snapshot };	71	return { _tag: "Success", bootstraps: snapshot };
72	}	72	}
73	try {	73	try {
74	snapshot = bridge	74	const next = bridge
75	.getLocalEnvironmentBootstraps()	75	.getLocalEnvironmentBootstraps()
76	.filter((entry) => entry.id !== PRIMARY_LOCAL_ENVIRONMENT_ID);	76	.filter((entry) => entry.id !== PRIMARY_LOCAL_ENVIRONMENT_ID);
		77	if (
		78	next.length !== snapshot.length \|\|
		79	next.some((entry, index) => {
		80	const previous = snapshot[index];
		81	return (
		82	previous === undefined \|\|
		83	entry.id !== previous.id \|\|
		84	entry.label !== previous.label \|\|
		85	entry.runningDistro !== previous.runningDistro \|\|
		86	entry.httpBaseUrl !== previous.httpBaseUrl \|\|
		87	entry.wsBaseUrl !== previous.wsBaseUrl \|\|
		88	entry.bootstrapToken !== previous.bootstrapToken
		89	);
		90	})
		91	) {
		92	snapshot = next;
		93	}
77	return { _tag: "Success", bootstraps: snapshot };	94	return { _tag: "Success", bootstraps: snapshot };
78	} catch (cause) {	95	} catch (cause) {
79	return { _tag: "Failure", cause };	96	return { _tag: "Failure", cause };

fix(orchestration): stabilize Codex turn mapping and settlement

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code