fix: coordinate task pull request sync consumers (#3090)

This commit is contained in:
Carlos Florêncio
2026-08-27 12:59:30 +00:00
committed by GitHub
parent d851ff7ca1
commit 6599a3b714
9 changed files with 1412 additions and 167 deletions
@@ -0,0 +1,368 @@
import { afterEach, beforeEach, describe, expect, it, vi } from "vitest";
import { createAppStore } from "@/lib/state/store";
import type { TaskPR } from "@/lib/types/github";
import {
createTaskPRSyncResource,
type TaskPRSyncRequester,
type TaskPRSyncScope,
} from "./task-pr-sync-resource";
function deferred<T>() {
let resolve!: (value: T) => void;
let reject!: (reason: unknown) => void;
const promise = new Promise<T>((resolvePromise, rejectPromise) => {
resolve = resolvePromise;
reject = rejectPromise;
});
return { promise, resolve, reject };
}
function createStore(workspaceId: string | null = "workspace-1", generation = 0) {
return createAppStore({
workspaces: { items: [], activeId: workspaceId },
workspaceContextGeneration: generation,
});
}
function makePR(overrides: Partial<TaskPR> = {}): TaskPR {
return {
id: "association-1",
workspace_id: "workspace-1",
task_id: "task-1",
owner: "acme",
repo: "app",
pr_number: 1,
pr_url: "https://github.com/acme/app/pull/1",
pr_title: "Task PR",
head_branch: "feature/task-1",
base_branch: "main",
author_login: "octocat",
state: "open",
review_state: "",
checks_state: "",
mergeable_state: "clean",
review_count: 0,
pending_review_count: 0,
comment_count: 0,
unresolved_review_threads: 0,
checks_total: 0,
checks_passing: 0,
additions: 0,
deletions: 0,
created_at: "2026-08-27T00:00:00Z",
merged_at: null,
closed_at: null,
last_synced_at: null,
updated_at: "2026-08-27T00:00:00Z",
...overrides,
};
}
const scope: TaskPRSyncScope = {
workspaceId: "workspace-1",
workspaceContextGeneration: 0,
taskId: "task-1",
};
beforeEach(() => {
vi.useFakeTimers();
});
afterEach(() => {
vi.useRealTimers();
});
describe("task PR sync resource request ownership", () => {
// @covers AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.1
// @covers AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.2
it("joins active requests and publishes one loaded snapshot to every lease", async () => {
const store = createStore();
const requester = vi.fn<TaskPRSyncRequester>();
const pending = deferred<{ prs: TaskPR[] }>();
requester.mockReturnValue(pending.promise);
const resource = createTaskPRSyncResource(store, requester);
const firstListener = vi.fn();
const secondListener = vi.fn();
const releaseFirst = resource.subscribe(scope, firstListener);
const releaseSecond = resource.subscribe(scope, secondListener);
const firstRefresh = resource.refresh(scope);
const secondRefresh = resource.refresh(scope);
expect(requester).toHaveBeenCalledTimes(1);
expect(firstRefresh).toBe(secondRefresh);
pending.resolve({ prs: [] });
await firstRefresh;
expect(resource.getSnapshot(scope)).toBe(true);
expect(firstListener).toHaveBeenCalled();
expect(secondListener).toHaveBeenCalled();
releaseFirst();
releaseSecond();
});
// @covers AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.4
it("keeps retries after one release and stops them after the final release", async () => {
const store = createStore();
const requester = vi.fn<TaskPRSyncRequester>().mockResolvedValue({ prs: [] });
const resource = createTaskPRSyncResource(store, requester, {
retryDelayMs: 10,
maxRetries: 3,
});
const releaseFirst = resource.subscribe(scope, vi.fn());
const releaseSecond = resource.subscribe(scope, vi.fn());
await Promise.resolve();
await vi.advanceTimersByTimeAsync(10);
expect(requester).toHaveBeenCalledTimes(2);
releaseFirst();
await vi.advanceTimersByTimeAsync(10);
expect(requester).toHaveBeenCalledTimes(3);
releaseSecond();
await vi.advanceTimersByTimeAsync(100);
expect(requester).toHaveBeenCalledTimes(3);
});
// @covers AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.3
it("keeps task, workspace, generation, and store scopes independent", async () => {
const store = createStore();
const requester = vi.fn<TaskPRSyncRequester>().mockResolvedValue({ prs: [] });
const resource = createTaskPRSyncResource(store, requester, { retryDelayMs: 100 });
const scopes = [
scope,
{ ...scope, taskId: "task-2" },
{ ...scope, workspaceId: "workspace-2" },
{ ...scope, workspaceContextGeneration: 1 },
];
const releases = scopes.map((item) => resource.subscribe(item, vi.fn()));
await Promise.resolve();
expect(requester).toHaveBeenCalledTimes(4);
releases.forEach((release) => release());
const secondStore = createStore();
const secondResource = createTaskPRSyncResource(secondStore, requester, {
retryDelayMs: 100,
});
const release = secondResource.subscribe(scope, vi.fn());
await Promise.resolve();
expect(requester).toHaveBeenCalledTimes(5);
release();
});
it("allows a Strict Mode-style release and reacquisition to rejoin the request", async () => {
const store = createStore();
const requester = vi.fn<TaskPRSyncRequester>();
const pending = deferred<{ prs: TaskPR[] }>();
requester.mockReturnValue(pending.promise);
const resource = createTaskPRSyncResource(store, requester);
const releaseFirst = resource.subscribe(scope, vi.fn());
releaseFirst();
const releaseSecond = resource.subscribe(scope, vi.fn());
expect(requester).toHaveBeenCalledTimes(1);
pending.resolve({ prs: [] });
await resource.refresh(scope);
releaseSecond();
});
});
describe("task PR sync resource lifecycle", () => {
// @covers AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.5
it("stops retrying after a permanent response", async () => {
const store = createStore();
const requester = vi.fn<TaskPRSyncRequester>().mockResolvedValue({
prs: [],
permanent: true,
});
const resource = createTaskPRSyncResource(store, requester, { retryDelayMs: 10 });
const release = resource.subscribe(scope, vi.fn());
await Promise.resolve();
await vi.advanceTimersByTimeAsync(100);
expect(requester).toHaveBeenCalledTimes(1);
expect(resource.getSnapshot(scope)).toBe(true);
release();
});
it("publishes discovered pull requests to the scoped store", async () => {
const store = createStore();
const requester = vi.fn<TaskPRSyncRequester>().mockResolvedValue({ prs: [makePR()] });
const resource = createTaskPRSyncResource(store, requester, { retryDelayMs: 10 });
const release = resource.subscribe(scope, vi.fn());
await Promise.resolve();
expect(store.getState().taskPRs.byTaskId[scope.taskId]).toEqual([makePR()]);
expect(resource.getSnapshot(scope)).toBe(true);
await vi.advanceTimersByTimeAsync(100);
expect(requester).toHaveBeenCalledTimes(1);
release();
});
it("marks a task loaded only after the final transient failure", async () => {
const store = createStore();
const requester = vi.fn<TaskPRSyncRequester>().mockRejectedValue(new Error("sync unavailable"));
const resource = createTaskPRSyncResource(store, requester, {
retryDelayMs: 10,
maxRetries: 2,
});
const release = resource.subscribe(scope, vi.fn());
await Promise.resolve();
expect(resource.getSnapshot(scope)).toBe(false);
await vi.advanceTimersByTimeAsync(10);
expect(resource.getSnapshot(scope)).toBe(false);
await vi.advanceTimersByTimeAsync(10);
expect(resource.getSnapshot(scope)).toBe(true);
expect(requester).toHaveBeenCalledTimes(3);
await vi.advanceTimersByTimeAsync(100);
expect(requester).toHaveBeenCalledTimes(3);
release();
});
it("anchors the retry delay to the start of the previous request", async () => {
const store = createStore();
const requester = vi.fn<TaskPRSyncRequester>();
const first = deferred<{ prs: TaskPR[] }>();
const second = deferred<{ prs: TaskPR[] }>();
requester.mockReturnValueOnce(first.promise).mockReturnValueOnce(second.promise);
const resource = createTaskPRSyncResource(store, requester, { retryDelayMs: 10 });
const release = resource.subscribe(scope, vi.fn());
expect(requester).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(8);
first.resolve({ prs: [] });
await vi.advanceTimersByTimeAsync(0);
await vi.advanceTimersByTimeAsync(1);
expect(requester).toHaveBeenCalledTimes(1);
await vi.advanceTimersByTimeAsync(1);
expect(requester).toHaveBeenCalledTimes(2);
second.resolve({ prs: [] });
await vi.advanceTimersByTimeAsync(0);
release();
});
it("stops the retry schedule when another path discovers a pull request", async () => {
const store = createStore();
const requester = vi.fn<TaskPRSyncRequester>().mockResolvedValue({ prs: [] });
const resource = createTaskPRSyncResource(store, requester, { retryDelayMs: 10 });
const release = resource.subscribe(scope, vi.fn());
await Promise.resolve();
store.getState().setTaskPR("task-1", makePR(), {
workspaceId: scope.workspaceId,
workspaceContextGeneration: scope.workspaceContextGeneration,
});
await vi.advanceTimersByTimeAsync(100);
expect(requester).toHaveBeenCalledTimes(1);
release();
});
});
describe("task PR sync resource recovery", () => {
// @covers AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.5
it("starts one fresh retry window on the connected transition", async () => {
const store = createStore();
const requester = vi.fn<TaskPRSyncRequester>().mockResolvedValue({ prs: [] });
const resource = createTaskPRSyncResource(store, requester, {
retryDelayMs: 10,
maxRetries: 2,
});
const release = resource.subscribe(scope, vi.fn());
await Promise.resolve();
await vi.advanceTimersByTimeAsync(30);
const callsAfterExhaustion = requester.mock.calls.length;
store.getState().setConnectionStatus("connected", null);
await Promise.resolve();
expect(requester.mock.calls.length).toBe(callsAfterExhaustion + 1);
store.getState().setConnectionStatus("connected", null);
await Promise.resolve();
expect(requester.mock.calls.length).toBe(callsAfterExhaustion + 1);
await vi.advanceTimersByTimeAsync(10);
expect(requester.mock.calls.length).toBe(callsAfterExhaustion + 2);
release();
});
it("abandons a pending pre-reconnect request and refreshes immediately", async () => {
const store = createStore();
const requester = vi.fn<TaskPRSyncRequester>();
const first = deferred<{ prs: TaskPR[] }>();
const second = deferred<{ prs: TaskPR[] }>();
requester.mockReturnValueOnce(first.promise).mockReturnValueOnce(second.promise);
const resource = createTaskPRSyncResource(store, requester);
const release = resource.subscribe(scope, vi.fn());
expect(requester).toHaveBeenCalledTimes(1);
store.getState().setConnectionStatus("connected", null);
expect(requester).toHaveBeenCalledTimes(2);
second.resolve({ prs: [] });
first.resolve({ prs: [makePR()] });
await vi.advanceTimersByTimeAsync(0);
expect(store.getState().taskPRs.byTaskId[scope.taskId]).toBeUndefined();
release();
});
// @covers AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.6
it("rejects a response from an inactive workspace context", async () => {
const store = createStore();
const requester = vi.fn<TaskPRSyncRequester>();
const pending = deferred<{ prs: TaskPR[] }>();
requester.mockReturnValue(pending.promise);
const resource = createTaskPRSyncResource(store, requester);
const release = resource.subscribe(scope, vi.fn());
store.setState((state) => ({
...state,
workspaces: { ...state.workspaces, activeId: "workspace-2" },
workspaceContextGeneration: 1,
}));
pending.resolve({ prs: [makePR()] });
await resource.refresh(scope);
expect(store.getState().taskPRs.byTaskId[scope.taskId]).toBeUndefined();
expect(resource.getSnapshot(scope)).toBe(false);
release();
});
it("does not publish an unresolved response after the final release", async () => {
const store = createStore();
const requester = vi.fn<TaskPRSyncRequester>();
const pending = deferred<{ prs: TaskPR[] }>();
requester.mockReturnValueOnce(pending.promise).mockResolvedValue({ prs: [] });
const resource = createTaskPRSyncResource(store, requester);
const release = resource.subscribe(scope, vi.fn());
release();
pending.resolve({ prs: [makePR()] });
await resource.refresh(scope);
expect(store.getState().taskPRs.byTaskId[scope.taskId]).toBeUndefined();
expect(resource.getSnapshot(scope)).toBe(false);
const reacquired = resource.subscribe(scope, vi.fn());
await Promise.resolve();
expect(requester).toHaveBeenCalledTimes(2);
reacquired();
});
});
@@ -0,0 +1,367 @@
import type { StoreApi } from "zustand";
import { getWebSocketClient } from "@/lib/ws/connection";
import type { AppState } from "@/lib/state/store";
import { isCurrentWorkspaceContext } from "@/lib/state/workspace-context";
import type { TaskPRScope } from "@/lib/state/slices/github/types";
import type { TaskPR } from "@/lib/types/github";
export type TaskPRSyncScope = TaskPRScope & {
taskId: string;
};
export type TaskPRSyncResponse =
| { prs?: TaskPR[]; permanent?: boolean }
| TaskPR
| null
| undefined;
export type TaskPRSyncRequester = (scope: TaskPRSyncScope) => Promise<TaskPRSyncResponse>;
export type TaskPRSyncResourceOptions = {
retryDelayMs?: number;
maxRetries?: number;
};
export type TaskPRSyncResource = {
getSnapshot: (scope: TaskPRSyncScope) => boolean;
invalidate: (scope: TaskPRSyncScope) => void;
refresh: (scope: TaskPRSyncScope) => Promise<void>;
subscribe: (scope: TaskPRSyncScope, listener: () => void) => () => void;
};
const DEFAULT_RETRY_DELAY_MS = 5_000;
const DEFAULT_MAX_RETRIES = 6;
const NO_WEBSOCKET_CLIENT = Symbol();
type ResourceEntry = {
key: string;
scope: TaskPRSyncScope;
loaded: boolean;
listeners: Set<() => void>;
promise: Promise<void> | null;
requestStartedAt: number | null;
requestGeneration: number;
retryCount: number;
permanent: boolean;
retryTimer: ReturnType<typeof setTimeout> | null;
stopStoreSubscription: (() => void) | null;
connectionStatus: AppState["connection"]["status"];
};
type ResourceStore = {
appStore: StoreApi<AppState>;
entries: Map<string, ResourceEntry>;
requester: TaskPRSyncRequester;
retryDelayMs: number;
maxRetries: number;
};
async function requestTaskPRSync(scope: TaskPRSyncScope): Promise<TaskPRSyncResponse> {
const client = getWebSocketClient();
if (!client) throw NO_WEBSOCKET_CLIENT;
return client.request<TaskPRSyncResponse>("github.task_pr.sync", { task_id: scope.taskId });
}
function scopeKey(scope: TaskPRSyncScope): string {
// The NUL delimiter is an in-memory scope boundary; API IDs are opaque
// stable values and are not expected to contain it.
return `${scope.workspaceId ?? ""}\u0000${scope.workspaceContextGeneration}\u0000${scope.taskId}`;
}
function createEntry(scope: TaskPRSyncScope): ResourceEntry {
return {
key: scopeKey(scope),
scope,
loaded: false,
listeners: new Set(),
promise: null,
requestStartedAt: null,
requestGeneration: 0,
retryCount: 0,
permanent: false,
retryTimer: null,
stopStoreSubscription: null,
connectionStatus: "disconnected",
};
}
function entryFor(store: ResourceStore, scope: TaskPRSyncScope): ResourceEntry {
const key = scopeKey(scope);
const existing = store.entries.get(key);
if (existing) return existing;
const entry = createEntry(scope);
store.entries.set(key, entry);
return entry;
}
function isCurrentEntry(store: ResourceStore, entry: ResourceEntry): boolean {
return store.entries.get(entry.key) === entry;
}
function isCurrentContext(store: ResourceStore, entry: ResourceEntry): boolean {
return isCurrentWorkspaceContext(
storeState(store),
entry.scope.workspaceId,
entry.scope.workspaceContextGeneration,
);
}
function storeState(store: ResourceStore): AppState {
return store.appStore.getState();
}
function isActiveEntry(store: ResourceStore, entry: ResourceEntry): boolean {
return entry.listeners.size > 0 && isCurrentEntry(store, entry) && isCurrentContext(store, entry);
}
function hasTaskPRs(state: AppState, scope: TaskPRSyncScope): boolean {
return (
state.taskPRs.workspaceId === scope.workspaceId &&
state.taskPRs.workspaceContextGeneration === scope.workspaceContextGeneration &&
Array.isArray(state.taskPRs.byTaskId[scope.taskId]) &&
state.taskPRs.byTaskId[scope.taskId].length > 0
);
}
function notify(entry: ResourceEntry): void {
for (const listener of entry.listeners) listener();
}
function setLoaded(entry: ResourceEntry, loaded: boolean): void {
if (entry.loaded === loaded) return;
entry.loaded = loaded;
notify(entry);
}
function clearRetryTimer(entry: ResourceEntry): void {
if (entry.retryTimer === null) return;
clearTimeout(entry.retryTimer);
entry.retryTimer = null;
}
function canScheduleRetry(store: ResourceStore, entry: ResourceEntry): boolean {
return (
isActiveEntry(store, entry) &&
!entry.permanent &&
entry.retryCount < store.maxRetries &&
!entry.retryTimer
);
}
function scheduleRetry(
store: ResourceStore,
entry: ResourceEntry,
delayMs = store.retryDelayMs,
): void {
if (!canScheduleRetry(store, entry)) return;
const timer = setTimeout(() => {
if (entry.retryTimer !== timer) return;
entry.retryTimer = null;
if (!isActiveEntry(store, entry) || entry.permanent || entry.retryCount >= store.maxRetries) {
return;
}
entry.retryCount += 1;
void startRequest(store, entry);
}, delayMs);
entry.retryTimer = timer;
}
function normalizeSyncResponse(result: TaskPRSyncResponse): TaskPR[] {
if (!result) return [];
const envelope = result as { prs?: TaskPR[] };
if (Array.isArray(envelope.prs)) return envelope.prs;
const single = result as TaskPR;
return single.task_id ? [single] : [];
}
function canPublishRequest(
store: ResourceStore,
entry: ResourceEntry,
requestGeneration: number,
): boolean {
return entry.requestGeneration === requestGeneration && isActiveEntry(store, entry);
}
function publishResponse(
store: ResourceStore,
entry: ResourceEntry,
requestGeneration: number,
result: TaskPRSyncResponse,
): void {
if (!canPublishRequest(store, entry, requestGeneration)) return;
const permanent = Boolean((result as { permanent?: boolean } | null)?.permanent);
const list = normalizeSyncResponse(result);
if (permanent) {
entry.permanent = true;
entry.retryCount = store.maxRetries;
clearRetryTimer(entry);
}
setLoaded(entry, true);
if (list.length === 0) return;
const state = storeState(store);
if (
!isCurrentWorkspaceContext(
state,
entry.scope.workspaceId,
entry.scope.workspaceContextGeneration,
)
) {
return;
}
for (const pr of list) {
if (pr.task_id) store.appStore.getState().setTaskPR(entry.scope.taskId, pr, entry.scope);
}
entry.retryCount = 0;
clearRetryTimer(entry);
}
function publishFailure(
store: ResourceStore,
entry: ResourceEntry,
requestGeneration: number,
): void {
if (!canPublishRequest(store, entry, requestGeneration)) return;
if (entry.retryCount >= store.maxRetries) setLoaded(entry, true);
}
async function runRequest(
store: ResourceStore,
entry: ResourceEntry,
requestGeneration: number,
): Promise<void> {
let result: TaskPRSyncResponse;
try {
result = await store.requester(entry.scope);
} catch (error) {
if (error !== NO_WEBSOCKET_CLIENT) publishFailure(store, entry, requestGeneration);
return;
}
publishResponse(store, entry, requestGeneration, result);
}
function settleRequest(store: ResourceStore, entry: ResourceEntry, promise: Promise<void>): void {
if (entry.promise !== promise) return;
const requestStartedAt = entry.requestStartedAt;
entry.promise = null;
entry.requestStartedAt = null;
if (!isCurrentEntry(store, entry)) return;
if (entry.listeners.size === 0) {
store.entries.delete(entry.key);
return;
}
const state = storeState(store);
if (hasTaskPRs(state, entry.scope)) clearRetryTimer(entry);
else {
const retryDelay =
requestStartedAt === null
? store.retryDelayMs
: Math.max(0, requestStartedAt + store.retryDelayMs - Date.now());
scheduleRetry(store, entry, retryDelay);
}
}
function startRequest(store: ResourceStore, entry: ResourceEntry): Promise<void> {
if (entry.promise) return entry.promise;
const requestGeneration = ++entry.requestGeneration;
entry.requestStartedAt = Date.now();
const promise = runRequest(store, entry, requestGeneration);
entry.promise = promise;
void promise.then(
() => settleRequest(store, entry, promise),
() => settleRequest(store, entry, promise),
);
return promise;
}
function invalidate(store: ResourceStore, scope: TaskPRSyncScope): void {
const entry = store.entries.get(scopeKey(scope));
if (!entry) return;
entry.requestGeneration += 1;
}
function observeStore(store: ResourceStore, entry: ResourceEntry): void {
entry.connectionStatus = storeState(store).connection.status;
entry.stopStoreSubscription = store.appStore.subscribe((state) => {
if (!isActiveEntry(store, entry)) return;
const previousStatus = entry.connectionStatus;
entry.connectionStatus = state.connection.status;
if (state.connection.status === "connected" && previousStatus !== "connected") {
entry.retryCount = 0;
entry.permanent = false;
clearRetryTimer(entry);
entry.promise = null;
entry.requestStartedAt = null;
void startRequest(store, entry);
return;
}
if (hasTaskPRs(state, entry.scope)) clearRetryTimer(entry);
else if (!entry.promise) scheduleRetry(store, entry);
});
}
function releaseEntry(store: ResourceStore, entry: ResourceEntry): void {
clearRetryTimer(entry);
entry.stopStoreSubscription?.();
entry.stopStoreSubscription = null;
if (entry.promise) return;
if (isCurrentEntry(store, entry)) store.entries.delete(entry.key);
}
function subscribe(store: ResourceStore, scope: TaskPRSyncScope, listener: () => void): () => void {
const entry = entryFor(store, scope);
entry.listeners.add(listener);
if (entry.listeners.size === 1) {
observeStore(store, entry);
if (!entry.promise) void startRequest(store, entry);
}
let released = false;
return () => {
if (released) return;
released = true;
entry.listeners.delete(listener);
if (entry.listeners.size > 0) return;
releaseEntry(store, entry);
};
}
function createResource(
store: StoreApi<AppState>,
requester: TaskPRSyncRequester,
options: TaskPRSyncResourceOptions,
): TaskPRSyncResource {
const resourceStore: ResourceStore = {
appStore: store,
entries: new Map(),
requester,
retryDelayMs: Math.max(0, options.retryDelayMs ?? DEFAULT_RETRY_DELAY_MS),
maxRetries: Math.max(0, options.maxRetries ?? DEFAULT_MAX_RETRIES),
};
return {
getSnapshot: (scope) => resourceStore.entries.get(scopeKey(scope))?.loaded ?? false,
invalidate: (scope) => invalidate(resourceStore, scope),
refresh: (scope) => startRequest(resourceStore, entryFor(resourceStore, scope)),
subscribe: (scope, listener) => subscribe(resourceStore, scope, listener),
};
}
export function createTaskPRSyncResource(
store: StoreApi<AppState>,
requester: TaskPRSyncRequester = requestTaskPRSync,
options: TaskPRSyncResourceOptions = {},
): TaskPRSyncResource {
return createResource(store, requester, options);
}
const resourcesByStore = new WeakMap<StoreApi<AppState>, TaskPRSyncResource>();
export function getTaskPRSyncResource(store: StoreApi<AppState>): TaskPRSyncResource {
const existing = resourcesByStore.get(store);
if (existing) return existing;
const resource = createTaskPRSyncResource(store);
resourcesByStore.set(store, resource);
return resource;
}
@@ -40,6 +40,23 @@ function useTaskPRWithStore(taskId: string | null) {
return { result, store };
}
function useTwoTaskPRs(taskId: string | null) {
return {
first: useTaskPR(taskId),
second: useTaskPR(taskId),
};
}
function deferred<T>() {
let resolve!: (value: T) => void;
let reject!: (reason: unknown) => void;
const promise = new Promise<T>((resolvePromise, rejectPromise) => {
resolve = resolvePromise;
reject = rejectPromise;
});
return { promise, resolve, reject };
}
const linkedPR = { id: "association-1", task_id: "task-1" } as TaskPR;
function unlinkState(workspaceId: string | null): Partial<AppState> {
@@ -135,6 +152,31 @@ describe("useTaskPR — loading state", () => {
});
});
describe("useTaskPR — shared synchronization", () => {
// @covers AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.1
it("shares initial and retry requests across consumers in one store", async () => {
requestMock.mockResolvedValue({ prs: [] });
const { result } = renderHook(() => useTwoTaskPRs("task-1"), {
wrapper: createStateWrapper({
workspaces: { items: [], activeId: "workspace-1" },
}),
});
await act(async () => {
await vi.advanceTimersByTimeAsync(0);
});
expect(requestMock).toHaveBeenCalledTimes(1);
await act(async () => {
await vi.advanceTimersByTimeAsync(5_000);
});
expect(requestMock).toHaveBeenCalledTimes(2);
expect(result.current.first.loaded).toBe(true);
expect(result.current.second.loaded).toBe(true);
});
});
describe("useTaskPR — unlink", () => {
it("rejects without a task or active workspace and skips the API", async () => {
const noTask = renderHook(() => useTaskPR(null), {
@@ -167,6 +209,33 @@ describe("useTaskPR — unlink", () => {
expect(view.result.current.store.getState().taskPRs.byTaskId["task-1"]).toBeUndefined();
});
it("drops an in-flight sync response after unlink removes local state", async () => {
const pending = deferred<{ prs: TaskPR[] }>();
requestMock.mockReturnValueOnce(pending.promise);
deleteTaskPRMock.mockResolvedValue(undefined);
const view = renderHook(() => useTaskPRWithStore("task-1"), {
wrapper: createStateWrapper(unlinkState("ws-1")),
});
await act(async () => {
await vi.advanceTimersByTimeAsync(0);
});
expect(requestMock).toHaveBeenCalledTimes(1);
await act(async () => {
await view.result.current.result.unlink(linkedPR.id);
});
expect(view.result.current.store.getState().taskPRs.byTaskId["task-1"]).toBeUndefined();
pending.resolve({ prs: [linkedPR] });
await act(async () => {
await pending.promise;
await vi.advanceTimersByTimeAsync(0);
});
expect(view.result.current.store.getState().taskPRs.byTaskId["task-1"]).toBeUndefined();
});
it("propagates API failures without removing the local association", async () => {
deleteTaskPRMock.mockRejectedValue(new Error("unlink failed"));
const view = renderHook(() => useTaskPRWithStore("task-1"), {
+25 -167
View File
@@ -1,10 +1,10 @@
"use client";
import { useEffect, useCallback, useRef, useState } from "react";
import { useCallback, useEffect, useMemo, useRef, useSyncExternalStore } from "react";
import { deleteTaskPR, listWorkspaceTaskPRs } from "@/lib/api/domains/github-api";
import { getWebSocketClient } from "@/lib/ws/connection";
import { useAppStore } from "@/components/state-provider";
import { useAppStore, useAppStoreApi } from "@/components/state-provider";
import { getTaskPRsForCurrentWorkspace } from "./use-task-pr-tooltip-hydration";
import { getTaskPRSyncResource, type TaskPRSyncScope } from "./task-pr-sync-resource";
import type { TaskPR } from "@/lib/types/github";
/** Fetch all PR associations for a workspace. */
@@ -40,9 +40,6 @@ export function useWorkspacePRs(workspaceId: string | null) {
}, [workspaceContextGeneration, setTaskPRs, workspaceId]);
}
const SYNC_RETRY_DELAY = 5_000; // 5 seconds
const SYNC_MAX_RETRIES = 6; // Up to 30 seconds of retries
/**
* Returns the primary PR (first by created_at) for a task. Multi-repo tasks
* may have additional PRs — use `useTaskPRs` to get the full list.
@@ -51,190 +48,51 @@ export function getPrimaryTaskPR(prs: TaskPR[] | undefined): TaskPR | null {
return prs && prs.length > 0 ? prs[0] : null;
}
/**
* Normalises the WS sync response into an array of TaskPR rows. Backend
* returns `{prs: TaskPR[]}` (current shape) for multi-repo support, but we
* accept the legacy bare-TaskPR shape too in case an older backend is
* still running. Empty / null / unknown shapes return an empty array.
*/
function normalizeSyncResponse(result: SyncResponse): TaskPR[] {
if (!result) return [];
const envelope = result as { prs?: TaskPR[] };
if (Array.isArray(envelope.prs)) return envelope.prs;
const single = result as TaskPR;
if (single.task_id) return [single];
return [];
}
/**
* Response shape from the `github.task_pr.sync` WS action. `permanent`
* is true when every watch on the task points at a repository the
* backend has classified as unresolvable (missing, deleted, or
* inaccessible). When set, the 5s retry interval stops — without this
* the frontend kept hammering a dead repo every 5s for the lifetime of
* the task. Older backends omit the field, so it's optional.
*/
type SyncResponse = { prs?: TaskPR[]; permanent?: boolean } | TaskPR | null | undefined;
/** Fetch a single task's PR associations, with on-demand sync via WS. */
export function useTaskPR(taskId: string | null) {
const store = useAppStoreApi();
const prs = useAppStore((state) =>
taskId ? getTaskPRsForCurrentWorkspace(state, taskId) : null,
);
const pr = getPrimaryTaskPR(prs ?? undefined);
const setTaskPR = useAppStore((state) => state.setTaskPR);
const removeTaskPR = useAppStore((state) => state.removeTaskPR);
const workspaceId = useAppStore((state) => state.workspaces.activeId);
const workspaceContextGeneration = useAppStore((state) => state.workspaceContextGeneration);
const connectionStatus = useAppStore((state) => state.connection.status);
const retryRef = useRef(0);
const permanentRef = useRef(false);
// Tracks the previous connection status so the resync effect below only
// fires on the transition edge into `connected`, not on every re-render
// while already connected.
const prevConnectionStatusRef = useRef(connectionStatus);
// Bumped on each reconnect transition so the retry-polling effect below
// tears down and recreates its interval (see that effect's dependency list).
const [reconnectGeneration, setReconnectGeneration] = useState(0);
const [loadedTaskId, setLoadedTaskId] = useState<string | null>(null);
// Monotonic counter incremented before each WS request, snapshotted in
// the .then() closure. Mirrors useWorkspacePRs above. Without this, a
// stale response from a previous taskId can land after the user
// navigates to a new task and flip permanentRef.current = true for the
// active task, killing its retry loop. The reset effect below clears
// retry/permanent state on taskId change, but a still-in-flight WS
// call from the previous task can race that reset.
const requestRef = useRef(0);
// Resolves when the sync settles (or immediately when it can't be sent), so
// callers can show an "updating" affordance for exactly as long as the
// summary is actually stale. Never rejects — failures are absorbed below.
const refresh = useCallback((): Promise<void> => {
if (!taskId) return Promise.resolve();
const client = getWebSocketClient();
if (!client) return Promise.resolve();
// Backend returns `{prs: TaskPR[], permanent?: boolean}` — multi-repo
// tasks have one row per repo. We push each into the store so the
// per-repo PR icon stays in sync. Empty array means no watches yet
// (the freshness retry below handles the polling cadence). Legacy
// single-PR shape (`TaskPR` only) is detected via the absence of
// `.prs`. When `permanent` is true (every watch's repo is dead),
// exhaust the retry counter so the 5s interval below clears itself.
const requestId = ++requestRef.current;
const requestedTaskId = taskId;
return client
.request<SyncResponse>("github.task_pr.sync", { task_id: requestedTaskId })
.then((result) => {
// Drop responses that aren't the latest in-flight request for
// this hook instance — they'd otherwise corrupt permanentRef /
// retryRef for whatever task the user is now viewing. The
// taskId-change effect below bumps requestRef.current too, so
// requestId alone covers both stale-by-sequence and
// stale-by-task-change.
if (requestRef.current !== requestId) return;
const envelope = (result ?? {}) as { permanent?: boolean };
if (envelope.permanent) {
permanentRef.current = true;
retryRef.current = SYNC_MAX_RETRIES;
}
const list = normalizeSyncResponse(result);
setLoadedTaskId(requestedTaskId);
if (list.length === 0) return;
const scope = { workspaceId, workspaceContextGeneration };
for (const pr of list) {
if (pr.task_id) setTaskPR(requestedTaskId, pr, scope);
}
retryRef.current = 0;
})
.catch(() => {
if (requestRef.current !== requestId) return;
// Keep transient failures pending while retries remain, then settle the
// loading state so review-panel cleanup cannot be blocked forever.
if (retryRef.current >= SYNC_MAX_RETRIES) {
setLoadedTaskId(requestedTaskId);
}
});
}, [taskId, setTaskPR, workspaceContextGeneration, workspaceId]);
const resource = getTaskPRSyncResource(store);
const scope = useMemo<TaskPRSyncScope | null>(
() => (taskId ? { taskId, workspaceId, workspaceContextGeneration } : null),
[taskId, workspaceContextGeneration, workspaceId],
);
const subscribe = useCallback(
(listener: () => void) => (scope ? resource.subscribe(scope, listener) : () => undefined),
[resource, scope],
);
const getSnapshot = useCallback(
() => (scope ? resource.getSnapshot(scope) : false),
[resource, scope],
);
const resourceLoaded = useSyncExternalStore(subscribe, getSnapshot, getSnapshot);
const refresh = useCallback(
() => (scope ? resource.refresh(scope) : Promise.resolve()),
[resource, scope],
);
const unlink = useCallback(
async (associationId: string) => {
if (!taskId || !workspaceId) throw new Error("No active workspace is selected.");
await deleteTaskPR(associationId, workspaceId);
removeTaskPR(taskId, associationId, { workspaceId, workspaceContextGeneration });
if (scope) resource.invalidate(scope);
},
[removeTaskPR, taskId, workspaceContextGeneration, workspaceId],
[removeTaskPR, resource, scope, taskId, workspaceContextGeneration, workspaceId],
);
// Reset retry/permanent state when taskId changes. Bumping requestRef
// here invalidates any still-in-flight .then() closure from the prior
// taskId so it can't write to the new task's refs.
useEffect(() => {
retryRef.current = 0;
permanentRef.current = false;
requestRef.current++;
setLoadedTaskId(null);
}, [taskId]);
// Sync once when the task becomes active (freshness check).
// Intentionally excludes `pr` so WS-driven store updates don't re-trigger.
useEffect(() => {
if (!taskId) return;
refresh();
}, [taskId, refresh]);
// Retry polling when no PR is in the store yet. permanentRef short-circuits
// the interval entirely so a task whose repos are all dead doesn't tie up
// the backend's gh throttle on every 5s tick.
//
// Depends on `reconnectGeneration` so a reconnect (see the effect below,
// which zeroes retryRef/permanentRef and bumps the generation on the
// transition edge) tears down and recreates the interval, starting a fresh
// 30s retry window. Without that dep, once the budget is exhausted the
// interval clears itself and merely resetting the refs never recreates it —
// a single empty resync would then leave the store empty and the
// "Pull Request" tab missing.
useEffect(() => {
if (!taskId || pr || permanentRef.current) return;
const interval = setInterval(() => {
if (retryRef.current >= SYNC_MAX_RETRIES || permanentRef.current) {
clearInterval(interval);
return;
}
retryRef.current++;
refresh();
}, SYNC_RETRY_DELAY);
return () => clearInterval(interval);
}, [taskId, pr, refresh, reconnectGeneration]);
// Reconnect-driven resync. A task can be opened while the WS is down (e.g. a
// transient dev HMR disconnect, or a task auto-created by the PR-review
// watcher that the user navigates to directly). In that window the freshness
// sync and every 5s retry queue/time out with nothing, so the store never
// learns about the PR and the auto "Pull Request" tab is never offered. When
// the socket comes back up, clear the exhausted retry/permanent state, refire
// the sync, and bump reconnectGeneration so the polling effect above restarts
// its (possibly already self-cleared) interval and runs a fresh 30s retry
// window. Guarded on the transition edge into `connected` so unrelated store
// updates don't refire.
useEffect(() => {
const prev = prevConnectionStatusRef.current;
prevConnectionStatusRef.current = connectionStatus;
if (!taskId || connectionStatus !== "connected" || prev === "connected") return;
retryRef.current = 0;
permanentRef.current = false;
refresh();
setReconnectGeneration((g) => g + 1);
}, [taskId, connectionStatus, refresh]);
return {
pr,
prs: prs ?? [],
refresh,
unlink,
loaded: loadedTaskId === taskId || prs !== null,
loaded: resourceLoaded || prs !== null,
} as {
pr: TaskPR | null;
prs: TaskPR[];
@@ -0,0 +1,144 @@
---
created: 2026-08-27
updated: 2026-08-27
status: implemented
requirements:
- REQ-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001
system_design:
- ../../specs/integrations/system-design/github-task-pr-sync-coordination.md
legacy_specs: []
---
# Implementation Plan: GitHub task pull request sync coordination
## Overview
Replace per-hook task pull request synchronization state with one shared,
store-scoped resource per workspace context and task. The resource coordinates
the initial request, in-flight work, retry timer, reconnect edge, loaded state,
and cleanup while `useTaskPR` preserves its current public API.
The confirmed root cause is that every mounted `useTaskPR` instance owns its
own effects and refs. A task page can mount several consumers, so it sends
several identical `github.task_pr.sync` requests and creates several retry
schedules. Backend behavior is not the source of the duplication.
## Scope
### In scope
- Add a shared task PR sync resource keyed by store, workspace ID, workspace
generation, and task ID.
- Share automatic and manual in-flight synchronization requests.
- Give each scoped entry one retry timer and one reconnect observer.
- Reference-count consumers and clean up after the last release.
- Preserve task PR store writes, loaded outcomes, retries, reconnect recovery,
manual refresh, and unlink behavior.
- Add deterministic resource and hook regression tests.
### Out of scope
- Backend request coalescing.
- WebSocket protocol or GitHub provider changes.
- Changes to task PR rendering, copy, or interactions.
- Refactoring tooltip hydration, workspace-wide loading, or Git status
subscriptions.
## Technical approach
### Shared synchronization resource
Create `apps/web/hooks/domains/github/task-pr-sync-resource.ts` with a testable
factory. Keep registries in a `WeakMap` keyed by Zustand store. Key each entry by
workspace ID, workspace-context generation, and task ID.
Each entry exposes a stable snapshot and subscription for
`useSyncExternalStore`. The subscription owns a consumer lease. The first lease
starts freshness synchronization; the last release stops timers and observers
and invalidates stale completions.
The entry joins all active refresh calls to one promise. It owns one bounded
five-second retry schedule and detects one transition edge into `connected`.
Before publishing a result, it verifies the resource generation and active
workspace context.
### Hook integration
Refactor `useTaskPR` to obtain the current Zustand store API and scoped resource
entry. Replace local retry, permanent, request, reconnect, and loaded state with
the resource snapshot and delegated `refresh`.
Keep `pr` and `prs` selected from the current workspace's task PR slice. Keep
the existing `unlink` callback and return shape unchanged. No component caller
changes are expected.
## Test-first sequence
1. Add a failing multi-consumer hook regression. Two consumers for one scoped
task must send one initial request and one request on a retry tick.
2. Add failing resource tests for concurrent manual refresh, consumer release,
separate scope keys, reconnect recovery, permanent responses, retry
exhaustion, and stale workspace completion.
3. Implement the shared resource and integrate `useTaskPR` until the focused
tests pass.
4. Run the existing hook suite to prove unlink, loading, retry, and reconnect
behavior remains unchanged.
## Tests
- `AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.1` and `.2`: two
consumers share initial, retry, and concurrent manual refresh requests.
- `.3`: different stores, tasks, workspaces, and workspace generations use
separate entries.
- `.4`: one unmount preserves the shared lifecycle; the last unmount cancels
future retries and observers.
- `.5`: empty, transient, permanent, discovered-PR, exhausted, and reconnect
cases retain their current outcomes.
- `.6`: a stale context response cannot write associations or publish loaded
state for the active context.
- `.7`: all consumers observe the same loaded result; existing unlink tests
remain green.
## Verification commands
Run from the repository root unless a command changes directory:
```bash
(cd apps/web && pnpm exec vitest run hooks/domains/github/task-pr-sync-resource.test.ts hooks/domains/github/use-task-pr.test.tsx)
(cd apps/web && pnpm run typecheck)
(cd apps && pnpm --filter @kandev/web lint)
git diff --check
```
If `apps/node_modules` is absent in a fresh worktree, first run
`cd apps && pnpm install --frozen-lockfile`.
## Work orders
- [x] [Task 01: Coordinate task PR sync consumers](task-01-coordinate-task-pr-sync-consumers.md)
## Implementation waves and parallel candidates
Wave 1 is sequential:
- Task 01 owns the resource, hook integration, and shared regression boundary.
There are no parallel-safe work orders. Splitting the resource from the hook
would require both tasks to edit the same lifecycle contract and tests.
## Risks
- React Strict Mode can subscribe, release, and subscribe again during mount.
The resource must rejoin an unresolved entry instead of sending a second
freshness request.
- A workspace can be reloaded with the same ID but a new generation. Omitting
the generation from the key could apply stale synchronization state.
- Stopping only the in-flight request duplication would leave duplicate retry
schedules. The shared resource must own timers and reconnect transitions.
- Cleanup must not let one consumer stop retries required by another consumer.
- The change normalizes data ownership only. It must not alter desktop or
mobile rendering, touch targets, or navigation behavior.
## Open questions
None.
@@ -0,0 +1,150 @@
---
id: "01-coordinate-task-pr-sync-consumers"
title: "Coordinate task PR sync consumers"
status: done
wave: 1
depends_on: []
plan: "plan.md"
requirements:
- REQ-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001
acceptance_criteria:
- AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.1
- AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.2
- AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.3
- AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.4
- AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.5
- AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.6
- AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.7
system_design:
- ../../specs/integrations/system-design/github-task-pr-sync-coordination.md
---
# Task 01: Coordinate task PR sync consumers
## Summary
Add one shared synchronization resource for each store, workspace context, and
task. Refactor `useTaskPR` to lease that resource while preserving its current
return contract and task PR store behavior.
## In scope
- Add the task PR sync resource and its test factory.
- Share initial, retry, reconnect, and manual refresh requests.
- Publish one loaded snapshot to all consumers.
- Reference-count consumers and clean up timers, observers, and stale entries.
- Guard response writes by resource generation and active workspace context.
- Integrate the resource with `useTaskPR` without changing callers.
- Add focused resource and multi-consumer hook regression tests.
## Out of scope
- Backend or protocol changes.
- UI component changes.
- Task PR unlink or tooltip hydration redesign.
- Session Git status subscription deduplication.
## Acceptance
- Two mounted `useTaskPR` consumers for the same scoped task send one initial
request and one request on each retry tick.
- Concurrent manual refresh calls return the same active completion and send
one request.
- Different task or workspace-context keys remain independent.
- One consumer release preserves the lifecycle; the final release stops later
retries and prevents stale publication.
- Permanent results stop retries. Reconnect starts one immediate request and
one fresh bounded retry window.
- Empty results, transient failures, terminal failures, discovered pull
requests, loaded state, and unlink behavior match the current hook contract.
## TDD sequence
### Red
- Extend `use-task-pr.test.tsx` with two consumers under one `StateProvider` and
assert that the current implementation sends duplicate initial and retry
requests.
- Add `task-pr-sync-resource.test.ts` cases for request joining, lease cleanup,
scope isolation, reconnect transitions, retry terminal states, and stale
responses. Confirm they fail before the resource exists.
### Green
- Implement the smallest store-scoped resource that satisfies the tests.
- Delegate `useTaskPR` synchronization and loaded state to the resource.
- Keep store selectors and unlink behavior unchanged.
### Refactor
- Keep resource state transitions explicit and separate from the React hook.
- Reuse `isCurrentWorkspaceContext` and the existing TaskPR scope types.
- Remove the superseded per-hook retry, request, reconnect, and loaded refs and
effects.
## Verification
```bash
cd apps/web && pnpm exec vitest run hooks/domains/github/task-pr-sync-resource.test.ts hooks/domains/github/use-task-pr.test.tsx
cd apps/web && pnpm run typecheck
cd apps && pnpm --filter @kandev/web lint
git diff --check
```
## Files likely touched
- `apps/web/hooks/domains/github/task-pr-sync-resource.ts`
- `apps/web/hooks/domains/github/task-pr-sync-resource.test.ts`
- `apps/web/hooks/domains/github/use-task-pr.ts`
- `apps/web/hooks/domains/github/use-task-pr.test.tsx`
## Dependencies
None.
## Risks
- Strict Mode effect replay can expose cleanup and re-acquisition races.
- A stale request can write into a later workspace context if either the store
or generation guard is missing.
- Ref-count errors can stop a needed retry or keep an orphan timer alive.
- A promise-only solution can pass concurrent-call tests while still producing
duplicate requests on later retry ticks.
## Parallelism
`sequential`
The resource and hook integration share one lifecycle contract and one focused
regression suite.
## Inputs
- `REQ-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001` and its system design.
- The confirmed per-instance state in
`apps/web/hooks/domains/github/use-task-pr.ts`.
- The store-scoped request registry in
`use-task-pr-tooltip-hydration.ts` and external-resource pattern in
`pr-commits-resource.ts`.
- The repeated `github.task_pr.sync` requests in the reported browser trace.
## Output contract
Report the files changed, the RED failure, GREEN and regression test outcomes,
typecheck and lint results, and any remaining lifecycle risk in the primary
session. Keep the requirement, design, plan, and work-order statuses in sync.
## Results
- Added the store-scoped task PR sync resource and deterministic resource tests.
- Refactored `useTaskPR` to lease the resource while preserving its return
contract, loaded behavior, refresh operation, and unlink behavior.
- RED: the multi-consumer regression observed two initial requests from the
previous per-hook implementation instead of one.
- GREEN: the focused resource and hook suites pass with 21 tests.
- The web typecheck and lint pass, and no rendered or mobile interaction changed.
- The resource invalidates publication when its lease is gone or its workspace
context is no longer current. WebSocket request timeouts remain owned by the
existing client.
- Unlink invalidates the scoped request generation after local removal, so an
in-flight response cannot restore the removed association.
+2
View File
@@ -45,6 +45,7 @@ outcomes that expose those contracts.
- [Workspace GitHub Authentication](requirements/github-authentication.md)
- [GitHub PR Merge Queue](requirements/github-pr-merge-queue.md)
- [GitHub PR Merge Queue Recovery](requirements/github-pr-merge-queue-recovery.md)
- [GitHub task pull request sync coordination](requirements/github-task-pr-sync-coordination.md)
- [GitLab Integration](requirements/gitlab-integration.md)
- [GitLab MR Status Chip](requirements/gitlab-mr-status-chip.md)
- [GitLab MR Badge on the Sidebar and Tasks-List Rows](requirements/gitlab-mr-task-list-badges.md)
@@ -72,6 +73,7 @@ outcomes that expose those contracts.
- [Workspace GitHub Authentication System Design Part 3](system-design/github-authentication-03.md)
- [GitHub PR Merge Queue](system-design/github-pr-merge-queue.md)
- [GitHub PR Merge Queue Recovery](system-design/github-pr-merge-queue-recovery.md)
- [GitHub task pull request sync coordination](system-design/github-task-pr-sync-coordination.md)
- [GitLab Integration System Design Part 1](system-design/gitlab-integration-01.md)
- [GitLab Integration System Design Part 2](system-design/gitlab-integration-02.md)
- [GitLab MR Status Chip System Design Part 1](system-design/gitlab-mr-status-chip-01.md)
@@ -0,0 +1,104 @@
---
status: active
system: integrations
created: 2026-08-27
updated: 2026-08-27
owners:
- kandev
---
# GitHub task pull request sync coordination Requirements
## Overview
Several task surfaces can display the same GitHub pull request association at
the same time. Each surface currently starts its own freshness request, retry
schedule, and reconnect recovery for the same task. The duplicate requests do
not change the result, but they increase WebSocket traffic, backend work, and
diagnostic noise.
This requirement makes task pull request synchronization a shared operation
within one active workspace context while preserving the existing detection and
recovery behavior.
## Requirements
### REQ-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001: Coordinate task pull request synchronization
**Intent:** All mounted consumers of one task pull request association must
share one synchronization lifecycle for the same application store, workspace
context, and task.
#### Acceptance criteria
- **AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.1:** Concurrent
consumers in the same application store, workspace ID, workspace-context
generation, and task ID share one initial freshness request and one retry
schedule.
- **AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.2:** An automatic or
manual refresh joins the active request for that context. A refresh starts a
new request when no request is active.
- **AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.3:** Different task IDs,
workspace IDs, workspace-context generations, or application stores remain
independent and do not share requests or results.
- **AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.4:** Removing one
consumer does not interrupt the lifecycle used by remaining consumers.
Removing the last consumer stops future scheduled retries and releases the
context's connection observer.
- **AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.5:** Empty responses,
transient failures, permanent failures, retry exhaustion, pull request
discovery, and reconnect transitions retain the existing retry and recovery
outcomes, with only one owner for each context.
- **AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.6:** A response from an
inactive workspace context cannot update the active workspace's pull request
associations or synchronization state.
- **AC-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001.7:** Every consumer of a
shared context observes the same loaded state and the same store-backed pull
request associations. Existing manual refresh and unlink behavior remains
available through the task pull request hook.
## Scope
### In scope
- Frontend ownership of the `github.task_pr.sync` request lifecycle.
- Initial freshness synchronization, bounded retries, and reconnect recovery.
- Shared loading completion for all consumers of one task context.
- Cleanup when consumers or workspace contexts change.
### Out of scope
- Backend request coalescing or rate-limit policy changes.
- Changes to `github.task_pr.sync` payloads or responses.
- Pull request association persistence, unlink semantics, or presentation.
- Tooltip-only HTTP hydration and workspace-wide pull request loading.
- Session Git status subscription behavior.
## Regression scenarios
- **GIVEN** two mounted components request the same task in one workspace
context, **WHEN** the task has no pull request, **THEN** one initial request
and one request per retry tick are sent.
- **GIVEN** two consumers share a request, **WHEN** both request a manual refresh
before it settles, **THEN** both receive the same completion and only one
WebSocket request is sent.
- **GIVEN** two consumers share a retry lifecycle, **WHEN** one unmounts,
**THEN** retries continue for the remaining consumer; **WHEN** the last one
unmounts, **THEN** no later retry is sent.
- **GIVEN** a retry budget is exhausted or marked permanent, **WHEN** the
connection later transitions into connected, **THEN** one immediate resync
starts one fresh retry window.
- **GIVEN** an older workspace context has an unresolved request, **WHEN** the
active workspace context changes before that request settles, **THEN** the
older response does not update the active context.
## Verification
Focused frontend tests count outbound synchronization requests and observe
shared hook state across multiple consumers. They also cover context isolation,
consumer cleanup, manual refresh, retry exhaustion, permanent responses, and
reconnect recovery.
This change has no rendered or interaction change. Desktop and mobile surfaces
continue to consume the same hook contract, so no new viewport-specific E2E
scenario is required.
@@ -0,0 +1,183 @@
---
status: current
system: integrations
requirements:
- REQ-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001
created: 2026-08-27
owners:
- kandev
---
# GitHub task pull request sync coordination System Design
## Purpose and boundaries
This design gives `github.task_pr.sync` one frontend lifecycle owner for each
application store, workspace context, and task. It removes duplicate requests
from concurrently mounted `useTaskPR` consumers without changing the WebSocket
contract, pull request store, retry policy, unlink operation, or rendered UI.
The integration system owns this design because it coordinates provider
synchronization. The UI remains a consumer of the existing hook result.
## Requirement mapping
| Requirement | Design sections |
| --- | --- |
| `REQ-INTEGRATIONS-GITHUB-TASK-PR-SYNC-COORDINATION-001` | [Resource identity and ownership](#resource-identity-and-ownership), [Request and retry lifecycle](#request-and-retry-lifecycle), [Hook integration](#hook-integration), [Failure and recovery](#failure-and-recovery) |
## Current failure mode
`useTaskPR` is mounted by the PR top bar, status chip, detail panel, Dockview
headers and watermark, review-panel synchronization, task shortcut, and
external-file-link detection. Every hook instance currently owns independent
request, retry, permanent, reconnect, and loaded state.
Mounting several of these consumers for one task therefore sends several
`github.task_pr.sync` freshness requests. Empty results create several
five-second retry intervals, and reconnect transitions can restart all of them.
The requests are semantically identical.
An in-flight promise map alone is insufficient. It can collapse calls that
overlap in time, but independent intervals can issue new requests immediately
after the shared promise settles. The shared abstraction must own the complete
lifecycle.
## Resource identity and ownership
Add `apps/web/hooks/domains/github/task-pr-sync-resource.ts`. The module exposes
a testable resource factory and an application resource registry.
The registry is a `WeakMap<StoreApi<AppState>, ...>` so separate state providers
cannot share requests. Each store registry keys entries by:
```text
workspaceId + workspaceContextGeneration + taskId
```
The workspace-context generation prevents a response from an earlier load of
the same workspace ID from joining or updating the current context.
Each entry owns:
- the immutable task and workspace scope;
- the current `loaded` snapshot and snapshot listeners;
- the active request promise;
- retry count, permanent status, and one retry timer;
- the previous connection status and one store connection observer;
- a consumer lease count and a generation used to reject stale completion;
- cleanup state for an orphaned in-flight request.
The task pull request Zustand slice remains the source of truth for associated
pull requests. The resource stores only synchronization lifecycle state.
## Request and retry lifecycle
The first consumer lease for an entry starts one freshness request. Additional
leases subscribe to the same entry and do not start another request.
`refresh` returns the active promise when a request is already running. When no
request is active, it sends `github.task_pr.sync` with the scoped task ID. This
rule applies to automatic and manual refreshes.
Response handling preserves the current hook policy:
- A successful response marks the scoped task as loaded.
- A non-empty response writes each association through `setTaskPR` with the
entry's workspace scope, resets the retry count, and stops retry scheduling.
- An empty response leaves pull request detection eligible for the bounded
retry schedule.
- A permanent response stops the retry schedule.
- A transient request failure keeps loading pending while retry attempts
remain. The final failed retry marks the task as loaded.
- One retry is attempted every five seconds, up to six attempts after the
initial freshness request.
The entry observes scoped task pull request state in the application store. A
pull request added through any path cancels the retry timer. If the scoped
association later becomes empty while a consumer remains, the entry can resume
the remaining retry lifecycle as the current hook does.
The last consumer release cancels the retry timer, removes the connection and
task-state observer, and invalidates the entry generation. An unresolved
request can settle for its callers, but it cannot publish lifecycle state or
write pull requests after the entry becomes stale. The entry is removed after
settlement so a later mount performs a new freshness check.
## Connection recovery
The entry, rather than each hook, detects the transition edge into
`connected`. That edge clears the retry and permanent state, starts one
immediate refresh, and starts one fresh bounded retry window if the result is
empty.
Repeated writes of `connected` do nothing. Different scoped entries recover
independently.
## Hook integration
Refactor `apps/web/hooks/domains/github/use-task-pr.ts` to use
`useAppStoreApi` and the shared task PR sync resource.
`useSyncExternalStore` binds each hook instance to the resource's stable
`loaded` snapshot. Its subscription is also the consumer lease: the first
subscriber activates the entry and the last subscriber releases it. Strict
Mode unsubscribe and resubscribe cycles must rejoin an active request instead
of starting a duplicate request.
The hook continues to derive `pr` and `prs` from the scoped Zustand slice. It
continues to expose:
- `refresh`, delegated to the scoped resource;
- `unlink`, retaining the current workspace and generation guards;
- `loaded`, true when either the resource has settled the scoped load or the
scoped store already contains an association.
After a successful unlink, the hook invalidates the scoped resource generation
after removing the local association. A response already in flight can settle
for its caller, but it cannot restore the removed association.
Call sites do not change. All existing surfaces benefit from coordination
through the hook boundary.
## Failure and recovery
- If the WebSocket client is unavailable, the resource does not invent a
successful result. Existing reconnect recovery remains the path to a fresh
window.
- A request rejection is absorbed by `refresh`, matching the existing
non-rejecting hook contract.
- Before writing a response, the resource verifies both its entry generation
and `isCurrentWorkspaceContext` against the current store state.
- A task or workspace switch releases the old entry and acquires a distinct
entry. The old response cannot alter the new entry's loaded or permanent
state.
- The application registry is store-scoped and entries are released after the
last consumer, preventing cross-provider leaks and unbounded settled state.
## Test design
Add resource-level tests for request joining, retry ownership, manual refresh,
consumer leases, reconnect edges, permanent responses, terminal failures, and
stale-context rejection. Use injected request and timer options so tests remain
deterministic.
Extend `use-task-pr.test.tsx` with a multi-consumer regression that renders two
`useTaskPR` instances under one `StateProvider`. Assert one initial request, one
request per retry tick, shared loaded state, continued retries after one
consumer leaves, and cleanup after the last consumer leaves. Keep existing
unlink and reconnect tests green.
Separate task and workspace-context tests prove that keys do not over-deduplicate.
No new desktop or mobile E2E test is required because the hook return contract
and all rendered interactions remain unchanged.
## Related designs
- [GitHub task pull request sync coordination requirements](../requirements/github-task-pr-sync-coordination.md)
- [PR task status summary requirements](../../ui/requirements/pr-task-status-summary.md)
- `apps/web/hooks/domains/github/pr-commits-resource.ts`, the existing shared
external-resource pattern for GitHub PR commit requests
- `apps/web/hooks/domains/github/use-task-pr-tooltip-hydration.ts`, the existing
store-scoped, workspace-generation-aware request registry pattern