Scheduled runs each get their own workspace in the sidebar (#1934)
* feat(schedules): per-run workspaces, isolation and archive controls Scheduled agents never appeared in the sidebar: each schedule reused one stamped workspace and archived the run's agent on completion. Now every run mints its own workspace — the entity the sidebar shows — with per-schedule controls for isolation (local or worktree) and whether to archive that workspace when the run finishes. The schedule form is rebuilt on a non-React form model (open/commands/close) to end the flicker, edit-into-create contamination, and multi-host hydration bugs that came from effect-choreographed shared state. That lands the reusable foundations it needed: a form kit over a single control-geometry owner, a data-access taxonomy (replica/snapshot/fetch) with real load-state semantics, and lint/boundary guardrails so new screens inherit the paved road. Also fixes desktop combobox popovers truncating options when the trigger is narrower than the content. New schedule protocol fields are optional on both create and update, so old and new clients/daemons stay compatible. * test(app): complete mock themes for control-geometry tokens switch and terminal-profile-edit-modal now style through createControlGeometry, which reads theme.borderWidth[1], theme.opacity[50], and theme.colors.borderAccent. The hand-rolled mock themes in these two suites lacked those tokens, so the eager StyleSheet.create mock threw at import (collection error), not on any assertion. Add the missing tokens. * fix(schedules): address review findings on run cleanup, edit isolation, load error, sheet dismiss, and preference hydration - Server: archive the run's workspace even when agent creation fails, so a misconfigured provider no longer orphans a sidebar workspace/worktree. - Edit form: keep a stored worktree isolation while worktree eligibility is still resolving, instead of silently downgrading a saved schedule to local. - Schedules screen: show the load error/retry UI when every host fails, rather than spinning forever behind the loading branch. - Form sheet: fire the parent onClose on native gesture/backdrop dismiss so the sheet closes instead of re-opening. - Form model: apply late-loading saved preferences to untouched create-mode fields (never to edited schedules or user-modified fields). * fix(schedules): review batch — reconnect resubscribe, worktree prune, Android dismiss, cadence and provider edit safety - Push-router: re-send terminal and checkout-diff subscriptions after a reconnect; the old per-hook effect resubscribed on isConnected and the extracted router did not, so daemon restarts silently stopped terminals_changed pushes (root cause of the terminal-activity e2e failures). - Server: pass the run's source repo root when archiving worktree runs so the worktree is unregistered (git worktree remove/prune), not just deleted; also archive the workspace when agent creation fails even with archive-off (an agentless workspace has nothing to inspect). - Sheet: RN Modal onDismiss is iOS-only — notify dismiss once-per-close on the Android native path too; replace the JSDOM component test with a real Playwright dismiss spec and inline the trivial dismiss decision. - Form model: editing an interval schedule no longer rewrites its cadence to cron unless the cadence UI is touched (90-minute intervals have no cron equivalent); switching projects clears stale provider/model state and gates submit until the new host's snapshot resolves. * chore(lint): drop custom lint wrapper and boundary script, plain oxlint The import bans stay in .oxlintrc.json (oxlint-native no-restricted-imports). The receiver-sensitive checks the custom script enforced are not worth a parallel lint pipeline; if they come back it will be as oxlint rules. * fix(schedules): crash-safe run cleanup, capability-gated run options, resilient mutations - Persist the run's workspace/agent ids on the running-run record so a daemon crash mid-run can be recovered: restart archives the interrupted run's workspace under the same policy as normal cleanup (agentless always archives, otherwise archiveOnFinish is respected). - Hide Archive on finish (and omit archiveOnFinish/isolation from payloads) on hosts without workspaceMultiplicity — an old daemon ignores the fields, so offering the switch there would lie. - Optimistic pause/resume/delete no longer throw when the schedules cache holds a still-connecting entry alongside loaded data. - Mode field is gated on provider mode options, not a selected model, so model-less providers can pick their advertised modes.
This commit is contained in:
Mohamed Boudra
@@ -0,0 +1,3 @@
|
||||
export function daemonConfigQueryKey(serverId: string | null) {
|
||||
return ["daemon-config", serverId] as const;
|
||||
}
|
||||
@@ -0,0 +1,51 @@
|
||||
import { describe, expect, it } from "vitest";
|
||||
import {
|
||||
isProvidersSnapshotHomeScope,
|
||||
normalizeProvidersSnapshotCwd,
|
||||
providersSnapshotQueryKey,
|
||||
providersSnapshotQueryRoot,
|
||||
providersSnapshotRequestOptions,
|
||||
} from "@/data/providers-snapshot";
|
||||
|
||||
describe("providers snapshot query scope", () => {
|
||||
it("normalizes blank cwd values to the home scope", () => {
|
||||
expect(normalizeProvidersSnapshotCwd(undefined)).toBeNull();
|
||||
expect(normalizeProvidersSnapshotCwd(null)).toBeNull();
|
||||
expect(normalizeProvidersSnapshotCwd(" ")).toBeNull();
|
||||
expect(isProvidersSnapshotHomeScope("")).toBe(true);
|
||||
});
|
||||
|
||||
it("keeps home and workspace query keys separate under one server root", () => {
|
||||
expect(providersSnapshotQueryRoot("server-1")).toEqual(["providersSnapshot", "server-1"]);
|
||||
expect(providersSnapshotQueryKey("server-1")).toEqual([
|
||||
"providersSnapshot",
|
||||
"server-1",
|
||||
"home",
|
||||
]);
|
||||
expect(providersSnapshotQueryKey("server-1", "/repo-a")).toEqual([
|
||||
"providersSnapshot",
|
||||
"server-1",
|
||||
"cwd",
|
||||
"/repo-a",
|
||||
]);
|
||||
});
|
||||
|
||||
it("builds request options with cwd only for workspace scopes", () => {
|
||||
expect(providersSnapshotRequestOptions({ cwd: null, providers: ["codex"] })).toEqual({
|
||||
providers: ["codex"],
|
||||
});
|
||||
expect(providersSnapshotRequestOptions({ cwd: "/repo-a", providers: ["codex"] })).toEqual({
|
||||
cwd: "/repo-a",
|
||||
providers: ["codex"],
|
||||
});
|
||||
});
|
||||
|
||||
it("uses one query scope for Windows cwd values with either separator", () => {
|
||||
expect(normalizeProvidersSnapshotCwd("C:\\Users\\Ezekiel Bulver\\project")).toBe(
|
||||
"C:/Users/Ezekiel Bulver/project",
|
||||
);
|
||||
expect(providersSnapshotQueryKey("server-1", "C:\\Users\\Ezekiel Bulver\\project")).toEqual(
|
||||
providersSnapshotQueryKey("server-1", "C:/Users/Ezekiel Bulver/project"),
|
||||
);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,34 @@
|
||||
import type { AgentProvider } from "@getpaseo/protocol/agent-types";
|
||||
import { normalizeWorkspacePath } from "@/utils/workspace-identity";
|
||||
|
||||
export const PROVIDERS_SNAPSHOT_QUERY_ROOT = "providersSnapshot";
|
||||
|
||||
export function normalizeProvidersSnapshotCwd(cwd?: string | null): string | null {
|
||||
return normalizeWorkspacePath(cwd);
|
||||
}
|
||||
|
||||
export function providersSnapshotQueryRoot(serverId: string | null) {
|
||||
return [PROVIDERS_SNAPSHOT_QUERY_ROOT, serverId] as const;
|
||||
}
|
||||
|
||||
export function providersSnapshotQueryKey(serverId: string | null, cwd?: string | null) {
|
||||
const normalizedCwd = normalizeProvidersSnapshotCwd(cwd);
|
||||
return normalizedCwd
|
||||
? ([PROVIDERS_SNAPSHOT_QUERY_ROOT, serverId, "cwd", normalizedCwd] as const)
|
||||
: ([PROVIDERS_SNAPSHOT_QUERY_ROOT, serverId, "home"] as const);
|
||||
}
|
||||
|
||||
export function providersSnapshotRequestOptions(input: {
|
||||
cwd?: string | null;
|
||||
providers?: AgentProvider[];
|
||||
}) {
|
||||
const normalizedCwd = normalizeProvidersSnapshotCwd(input.cwd);
|
||||
return {
|
||||
...(normalizedCwd ? { cwd: normalizedCwd } : {}),
|
||||
...(input.providers ? { providers: input.providers } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
export function isProvidersSnapshotHomeScope(cwd?: string | null): boolean {
|
||||
return normalizeProvidersSnapshotCwd(cwd) === null;
|
||||
}
|
||||
@@ -0,0 +1,518 @@
|
||||
import { QueryClient, QueryObserver, skipToken } from "@tanstack/react-query";
|
||||
import { describe, expect, it } from "vitest";
|
||||
import type { MutableDaemonConfig, SessionOutboundMessage } from "@getpaseo/protocol/messages";
|
||||
import { checkoutDiffQueryKey } from "@/git/query-keys";
|
||||
import { buildTerminalsQueryKey } from "@/screens/workspace/terminals/state";
|
||||
import { daemonConfigQueryKey } from "@/data/daemon-config";
|
||||
import { providersSnapshotQueryKey } from "@/data/providers-snapshot";
|
||||
import {
|
||||
checkoutDiffPushRoute,
|
||||
invalidateServerDataQueriesAfterReconnect,
|
||||
mountServerDataPushRouter,
|
||||
workspaceTerminalsPushRoute,
|
||||
} from "@/data/push-router";
|
||||
|
||||
type ProvidersSnapshotUpdateMessage = Extract<
|
||||
SessionOutboundMessage,
|
||||
{ type: "providers_snapshot_update" }
|
||||
>;
|
||||
type CheckoutDiffUpdateMessage = Extract<SessionOutboundMessage, { type: "checkout_diff_update" }>;
|
||||
type SubscribeCheckoutDiffResponseMessage = Extract<
|
||||
SessionOutboundMessage,
|
||||
{ type: "subscribe_checkout_diff_response" }
|
||||
>;
|
||||
type StatusMessage = Extract<SessionOutboundMessage, { type: "status" }>;
|
||||
type TerminalsChangedMessage = Extract<SessionOutboundMessage, { type: "terminals_changed" }>;
|
||||
type RouterMessage =
|
||||
| ProvidersSnapshotUpdateMessage
|
||||
| CheckoutDiffUpdateMessage
|
||||
| SubscribeCheckoutDiffResponseMessage
|
||||
| StatusMessage
|
||||
| TerminalsChangedMessage;
|
||||
type RouterMessageType = RouterMessage["type"];
|
||||
type RouterHandler = (message: RouterMessage) => void;
|
||||
type RouterClient = Parameters<typeof mountServerDataPushRouter>[0]["client"];
|
||||
|
||||
const daemonConfig: MutableDaemonConfig = {
|
||||
mcp: { injectIntoAgents: true },
|
||||
browserTools: { enabled: false },
|
||||
providers: {},
|
||||
metadataGeneration: { providers: [] },
|
||||
autoArchiveAfterMerge: false,
|
||||
enableTerminalAgentHooks: false,
|
||||
appendSystemPrompt: "",
|
||||
};
|
||||
|
||||
function createFakeClient(config: { rejectCheckoutDiffSubscribe?: boolean } = {}): {
|
||||
client: RouterClient;
|
||||
emit: <K extends RouterMessageType>(message: Extract<RouterMessage, { type: K }>) => void;
|
||||
subscribeCheckoutDiffCalls: Array<{
|
||||
cwd: string;
|
||||
compare: { mode: "uncommitted" | "base"; baseRef?: string; ignoreWhitespace?: boolean };
|
||||
subscriptionId: string;
|
||||
}>;
|
||||
unsubscribeCheckoutDiffCalls: string[];
|
||||
subscribeTerminalCalls: Array<{ cwd: string; workspaceId?: string }>;
|
||||
unsubscribeTerminalCalls: Array<{ cwd: string; workspaceId?: string }>;
|
||||
} {
|
||||
const handlers: Record<RouterMessageType, RouterHandler[]> = {
|
||||
providers_snapshot_update: [],
|
||||
checkout_diff_update: [],
|
||||
subscribe_checkout_diff_response: [],
|
||||
status: [],
|
||||
terminals_changed: [],
|
||||
};
|
||||
const subscribeCheckoutDiffCalls: Array<{
|
||||
cwd: string;
|
||||
compare: { mode: "uncommitted" | "base"; baseRef?: string; ignoreWhitespace?: boolean };
|
||||
subscriptionId: string;
|
||||
}> = [];
|
||||
const unsubscribeCheckoutDiffCalls: string[] = [];
|
||||
const subscribeTerminalCalls: Array<{ cwd: string; workspaceId?: string }> = [];
|
||||
const unsubscribeTerminalCalls: Array<{ cwd: string; workspaceId?: string }> = [];
|
||||
|
||||
function on<K extends RouterMessageType>(
|
||||
type: K,
|
||||
handler: (message: Extract<RouterMessage, { type: K }>) => void,
|
||||
): () => void {
|
||||
const routerHandler: RouterHandler = (message) => {
|
||||
if (message.type === type) {
|
||||
handler(message as Extract<RouterMessage, { type: K }>);
|
||||
}
|
||||
};
|
||||
handlers[type].push(routerHandler);
|
||||
return () => {
|
||||
handlers[type] = handlers[type].filter((candidate) => candidate !== routerHandler);
|
||||
};
|
||||
}
|
||||
|
||||
function emit<K extends RouterMessageType>(message: Extract<RouterMessage, { type: K }>): void {
|
||||
for (const handler of handlers[message.type]) {
|
||||
handler(message);
|
||||
}
|
||||
}
|
||||
|
||||
return {
|
||||
client: {
|
||||
on,
|
||||
async subscribeCheckoutDiff(cwd, compare, requestOptions) {
|
||||
subscribeCheckoutDiffCalls.push({
|
||||
cwd,
|
||||
compare,
|
||||
subscriptionId: requestOptions.subscriptionId,
|
||||
});
|
||||
if (config.rejectCheckoutDiffSubscribe) {
|
||||
throw new Error("subscribe failed");
|
||||
}
|
||||
return {
|
||||
subscriptionId: requestOptions.subscriptionId,
|
||||
cwd,
|
||||
files: [],
|
||||
error: null,
|
||||
requestId: requestOptions.requestId ?? "subscribe-checkout-diff",
|
||||
};
|
||||
},
|
||||
unsubscribeCheckoutDiff(subscriptionId) {
|
||||
unsubscribeCheckoutDiffCalls.push(subscriptionId);
|
||||
},
|
||||
subscribeTerminals(subscription) {
|
||||
subscribeTerminalCalls.push(subscription);
|
||||
},
|
||||
unsubscribeTerminals(subscription) {
|
||||
unsubscribeTerminalCalls.push(subscription);
|
||||
},
|
||||
},
|
||||
emit,
|
||||
subscribeCheckoutDiffCalls,
|
||||
unsubscribeCheckoutDiffCalls,
|
||||
subscribeTerminalCalls,
|
||||
unsubscribeTerminalCalls,
|
||||
};
|
||||
}
|
||||
|
||||
function providerUpdate(generatedAt: string): ProvidersSnapshotUpdateMessage {
|
||||
return {
|
||||
type: "providers_snapshot_update",
|
||||
payload: {
|
||||
entries: [{ provider: "codex", status: "ready", enabled: true, models: [] }],
|
||||
generatedAt,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
describe("server data push router", () => {
|
||||
it("routes provider snapshot and daemon config payloads until detached", () => {
|
||||
const queryClient = new QueryClient();
|
||||
const fake = createFakeClient();
|
||||
const serverId = "server-1";
|
||||
const unmount = mountServerDataPushRouter({ client: fake.client, queryClient, serverId });
|
||||
|
||||
fake.emit(providerUpdate("2026-01-01T00:00:00.000Z"));
|
||||
fake.emit({
|
||||
type: "status",
|
||||
payload: { status: "daemon_config_changed", config: daemonConfig },
|
||||
});
|
||||
|
||||
expect(queryClient.getQueryData(providersSnapshotQueryKey(serverId))).toEqual({
|
||||
entries: [{ provider: "codex", status: "ready", enabled: true, models: [] }],
|
||||
generatedAt: "2026-01-01T00:00:00.000Z",
|
||||
requestId: "providers_snapshot_update",
|
||||
});
|
||||
expect(queryClient.getQueryData(daemonConfigQueryKey(serverId))).toEqual(daemonConfig);
|
||||
|
||||
unmount();
|
||||
fake.emit(providerUpdate("2026-01-01T00:00:01.000Z"));
|
||||
|
||||
expect(queryClient.getQueryData(providersSnapshotQueryKey(serverId))).toEqual({
|
||||
entries: [{ provider: "codex", status: "ready", enabled: true, models: [] }],
|
||||
generatedAt: "2026-01-01T00:00:00.000Z",
|
||||
requestId: "providers_snapshot_update",
|
||||
});
|
||||
});
|
||||
|
||||
it("subscribes active checkout diff queries and writes matching diff events", () => {
|
||||
const queryClient = new QueryClient();
|
||||
const fake = createFakeClient();
|
||||
const serverId = "server-1";
|
||||
const cwd = "/repo";
|
||||
const queryKey = checkoutDiffQueryKey(serverId, cwd, "base", "main", true);
|
||||
const subscriptionId = `checkoutDiff:${JSON.stringify(queryKey)}`;
|
||||
const observer = new QueryObserver(queryClient, {
|
||||
queryKey,
|
||||
queryFn: skipToken,
|
||||
enabled: true,
|
||||
gcTime: Infinity,
|
||||
staleTime: Infinity,
|
||||
meta: checkoutDiffPushRoute({
|
||||
enabled: true,
|
||||
serverId,
|
||||
subscriptionId,
|
||||
cwd,
|
||||
compare: { mode: "base", baseRef: "main", ignoreWhitespace: true },
|
||||
}),
|
||||
});
|
||||
const unsubscribeObserver = observer.subscribe(() => undefined);
|
||||
const unmount = mountServerDataPushRouter({ client: fake.client, queryClient, serverId });
|
||||
|
||||
expect(fake.subscribeCheckoutDiffCalls).toEqual([
|
||||
{
|
||||
cwd,
|
||||
compare: { mode: "base", baseRef: "main", ignoreWhitespace: true },
|
||||
subscriptionId,
|
||||
},
|
||||
]);
|
||||
|
||||
fake.emit({
|
||||
type: "subscribe_checkout_diff_response",
|
||||
payload: { subscriptionId, cwd, files: [], error: null, requestId: "diff-1" },
|
||||
});
|
||||
|
||||
expect(queryClient.getQueryData(queryKey)).toEqual({
|
||||
cwd,
|
||||
files: [],
|
||||
error: null,
|
||||
requestId: "diff-1",
|
||||
});
|
||||
|
||||
fake.emit({
|
||||
type: "checkout_diff_update",
|
||||
payload: { subscriptionId, cwd, files: [], error: null },
|
||||
});
|
||||
|
||||
expect(queryClient.getQueryData(queryKey)).toEqual({
|
||||
cwd,
|
||||
files: [],
|
||||
error: null,
|
||||
requestId: `subscription:${subscriptionId}`,
|
||||
});
|
||||
|
||||
unsubscribeObserver();
|
||||
|
||||
expect(fake.unsubscribeCheckoutDiffCalls).toEqual([subscriptionId]);
|
||||
|
||||
unmount();
|
||||
});
|
||||
|
||||
it("does not retry failed subscriptions on unrelated cache events", async () => {
|
||||
const queryClient = new QueryClient();
|
||||
const fake = createFakeClient({ rejectCheckoutDiffSubscribe: true });
|
||||
const serverId = "server-1";
|
||||
const cwd = "/repo";
|
||||
const queryKey = checkoutDiffQueryKey(serverId, cwd, "base", "main", true);
|
||||
const subscriptionId = `checkoutDiff:${JSON.stringify(queryKey)}`;
|
||||
const observer = new QueryObserver(queryClient, {
|
||||
queryKey,
|
||||
queryFn: skipToken,
|
||||
enabled: true,
|
||||
gcTime: Infinity,
|
||||
staleTime: Infinity,
|
||||
meta: checkoutDiffPushRoute({
|
||||
enabled: true,
|
||||
serverId,
|
||||
subscriptionId,
|
||||
cwd,
|
||||
compare: { mode: "base", baseRef: "main", ignoreWhitespace: true },
|
||||
}),
|
||||
});
|
||||
const unsubscribeObserver = observer.subscribe(() => undefined);
|
||||
const unmount = mountServerDataPushRouter({ client: fake.client, queryClient, serverId });
|
||||
|
||||
expect(fake.subscribeCheckoutDiffCalls).toHaveLength(1);
|
||||
|
||||
await Promise.resolve();
|
||||
await Promise.resolve();
|
||||
|
||||
queryClient.setQueryData(["unrelated"], "value");
|
||||
|
||||
expect(fake.subscribeCheckoutDiffCalls).toHaveLength(1);
|
||||
|
||||
unsubscribeObserver();
|
||||
unmount();
|
||||
});
|
||||
|
||||
it("subscribes active terminal queries and filters terminal pushes by workspace", () => {
|
||||
const queryClient = new QueryClient();
|
||||
const fake = createFakeClient();
|
||||
const serverId = "server-1";
|
||||
const cwd = "/repo";
|
||||
const workspaceId = "workspace-a";
|
||||
const queryKey = buildTerminalsQueryKey(serverId, cwd, workspaceId);
|
||||
const observer = new QueryObserver(queryClient, {
|
||||
queryKey,
|
||||
queryFn: skipToken,
|
||||
enabled: true,
|
||||
gcTime: Infinity,
|
||||
staleTime: Infinity,
|
||||
meta: workspaceTerminalsPushRoute({
|
||||
enabled: true,
|
||||
serverId,
|
||||
cwd,
|
||||
workspaceId,
|
||||
}),
|
||||
});
|
||||
const unsubscribeObserver = observer.subscribe(() => undefined);
|
||||
const unmount = mountServerDataPushRouter({ client: fake.client, queryClient, serverId });
|
||||
|
||||
expect(fake.subscribeTerminalCalls).toEqual([{ cwd, workspaceId }]);
|
||||
|
||||
fake.emit({
|
||||
type: "terminals_changed",
|
||||
payload: {
|
||||
cwd,
|
||||
terminals: [
|
||||
{ id: "terminal-a", name: "Main", workspaceId },
|
||||
{ id: "terminal-b", name: "Sibling", workspaceId: "workspace-b" },
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
expect(queryClient.getQueryData(queryKey)).toEqual({
|
||||
cwd,
|
||||
terminals: [{ id: "terminal-a", name: "Main", workspaceId }],
|
||||
requestId: expect.stringMatching(/^terminals-changed-/),
|
||||
});
|
||||
|
||||
unsubscribeObserver();
|
||||
|
||||
expect(fake.unsubscribeTerminalCalls).toEqual([{ cwd, workspaceId }]);
|
||||
|
||||
unmount();
|
||||
});
|
||||
|
||||
it("re-sends active push subscriptions after reconnect", () => {
|
||||
const queryClient = new QueryClient();
|
||||
const fake = createFakeClient();
|
||||
const serverId = "server-1";
|
||||
const cwd = "/repo";
|
||||
const workspaceId = "workspace-a";
|
||||
const checkoutDiffKey = checkoutDiffQueryKey(serverId, cwd, "base", "main", true);
|
||||
const checkoutDiffSubscriptionId = `checkoutDiff:${JSON.stringify(checkoutDiffKey)}`;
|
||||
const terminalKey = buildTerminalsQueryKey(serverId, cwd, workspaceId);
|
||||
const checkoutDiffObserver = new QueryObserver(queryClient, {
|
||||
queryKey: checkoutDiffKey,
|
||||
queryFn: skipToken,
|
||||
enabled: true,
|
||||
gcTime: Infinity,
|
||||
staleTime: Infinity,
|
||||
meta: checkoutDiffPushRoute({
|
||||
enabled: true,
|
||||
serverId,
|
||||
subscriptionId: checkoutDiffSubscriptionId,
|
||||
cwd,
|
||||
compare: { mode: "base", baseRef: "main", ignoreWhitespace: true },
|
||||
}),
|
||||
});
|
||||
const terminalObserver = new QueryObserver(queryClient, {
|
||||
queryKey: terminalKey,
|
||||
queryFn: skipToken,
|
||||
enabled: true,
|
||||
gcTime: Infinity,
|
||||
staleTime: Infinity,
|
||||
meta: workspaceTerminalsPushRoute({
|
||||
enabled: true,
|
||||
serverId,
|
||||
cwd,
|
||||
workspaceId,
|
||||
}),
|
||||
});
|
||||
const unsubscribeCheckoutDiffObserver = checkoutDiffObserver.subscribe(() => undefined);
|
||||
const unsubscribeTerminalObserver = terminalObserver.subscribe(() => undefined);
|
||||
const unmount = mountServerDataPushRouter({ client: fake.client, queryClient, serverId });
|
||||
const plainCheckoutDiffObserver = new QueryObserver(queryClient, {
|
||||
queryKey: checkoutDiffKey,
|
||||
queryFn: skipToken,
|
||||
enabled: true,
|
||||
gcTime: Infinity,
|
||||
staleTime: Infinity,
|
||||
});
|
||||
const plainTerminalObserver = new QueryObserver(queryClient, {
|
||||
queryKey: terminalKey,
|
||||
queryFn: skipToken,
|
||||
enabled: true,
|
||||
gcTime: Infinity,
|
||||
staleTime: Infinity,
|
||||
});
|
||||
const unsubscribePlainCheckoutDiffObserver = plainCheckoutDiffObserver.subscribe(
|
||||
() => undefined,
|
||||
);
|
||||
const unsubscribePlainTerminalObserver = plainTerminalObserver.subscribe(() => undefined);
|
||||
|
||||
invalidateServerDataQueriesAfterReconnect({ queryClient, serverId });
|
||||
|
||||
expect(fake.subscribeCheckoutDiffCalls).toEqual([
|
||||
{
|
||||
cwd,
|
||||
compare: { mode: "base", baseRef: "main", ignoreWhitespace: true },
|
||||
subscriptionId: checkoutDiffSubscriptionId,
|
||||
},
|
||||
{
|
||||
cwd,
|
||||
compare: { mode: "base", baseRef: "main", ignoreWhitespace: true },
|
||||
subscriptionId: checkoutDiffSubscriptionId,
|
||||
},
|
||||
]);
|
||||
expect(fake.subscribeTerminalCalls).toEqual([
|
||||
{ cwd, workspaceId },
|
||||
{ cwd, workspaceId },
|
||||
]);
|
||||
|
||||
fake.emit({
|
||||
type: "terminals_changed",
|
||||
payload: {
|
||||
cwd,
|
||||
terminals: [
|
||||
{ id: "terminal-a", name: "Main", workspaceId },
|
||||
{ id: "terminal-b", name: "Sibling", workspaceId: "workspace-b" },
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
expect(queryClient.getQueryData(terminalKey)).toEqual({
|
||||
cwd,
|
||||
terminals: [{ id: "terminal-a", name: "Main", workspaceId }],
|
||||
requestId: expect.stringMatching(/^terminals-changed-/),
|
||||
});
|
||||
|
||||
unsubscribePlainCheckoutDiffObserver();
|
||||
unsubscribePlainTerminalObserver();
|
||||
unsubscribeCheckoutDiffObserver();
|
||||
unsubscribeTerminalObserver();
|
||||
unmount();
|
||||
});
|
||||
|
||||
it("routes terminal pushes after another observer attaches without push metadata", () => {
|
||||
const queryClient = new QueryClient();
|
||||
const fake = createFakeClient();
|
||||
const serverId = "server-1";
|
||||
const cwd = "/repo";
|
||||
const workspaceId = "workspace-a";
|
||||
const queryKey = buildTerminalsQueryKey(serverId, cwd, workspaceId);
|
||||
const pushObserver = new QueryObserver(queryClient, {
|
||||
queryKey,
|
||||
queryFn: skipToken,
|
||||
enabled: true,
|
||||
gcTime: Infinity,
|
||||
staleTime: Infinity,
|
||||
meta: workspaceTerminalsPushRoute({
|
||||
enabled: true,
|
||||
serverId,
|
||||
cwd,
|
||||
workspaceId,
|
||||
}),
|
||||
});
|
||||
const unsubscribePushObserver = pushObserver.subscribe(() => undefined);
|
||||
const unmount = mountServerDataPushRouter({ client: fake.client, queryClient, serverId });
|
||||
expect(fake.subscribeTerminalCalls).toEqual([{ cwd, workspaceId }]);
|
||||
|
||||
const plainObserver = new QueryObserver(queryClient, {
|
||||
queryKey,
|
||||
queryFn: skipToken,
|
||||
enabled: true,
|
||||
gcTime: Infinity,
|
||||
staleTime: Infinity,
|
||||
});
|
||||
const unsubscribePlainObserver = plainObserver.subscribe(() => undefined);
|
||||
|
||||
fake.emit({
|
||||
type: "terminals_changed",
|
||||
payload: {
|
||||
cwd,
|
||||
terminals: [
|
||||
{
|
||||
id: "terminal-a",
|
||||
name: "Main",
|
||||
workspaceId,
|
||||
activity: { state: "idle", attentionReason: "needs_input", changedAt: 1 },
|
||||
},
|
||||
],
|
||||
},
|
||||
});
|
||||
|
||||
expect(queryClient.getQueryData(queryKey)).toEqual({
|
||||
cwd,
|
||||
terminals: [
|
||||
{
|
||||
id: "terminal-a",
|
||||
name: "Main",
|
||||
workspaceId,
|
||||
activity: { state: "idle", attentionReason: "needs_input", changedAt: 1 },
|
||||
},
|
||||
],
|
||||
requestId: expect.stringMatching(/^terminals-changed-/),
|
||||
});
|
||||
expect(fake.unsubscribeTerminalCalls).toEqual([]);
|
||||
|
||||
unsubscribePlainObserver();
|
||||
unsubscribePushObserver();
|
||||
unmount();
|
||||
});
|
||||
|
||||
it("invalidates only the reconnect-repair scopes for one server", () => {
|
||||
const queryClient = new QueryClient();
|
||||
const serverId = "server-1";
|
||||
const otherServerId = "server-2";
|
||||
const providerKey = providersSnapshotQueryKey(serverId);
|
||||
const daemonConfigKey = daemonConfigQueryKey(serverId);
|
||||
const diffKey = checkoutDiffQueryKey(serverId, "/repo", "uncommitted", undefined, false);
|
||||
const terminalKey = buildTerminalsQueryKey(serverId, "/repo", "workspace-a");
|
||||
const otherProviderKey = providersSnapshotQueryKey(otherServerId);
|
||||
|
||||
queryClient.setQueryData(providerKey, { entries: [], generatedAt: "now", requestId: "p" });
|
||||
queryClient.setQueryData(daemonConfigKey, daemonConfig);
|
||||
queryClient.setQueryData(diffKey, { cwd: "/repo", files: [], error: null, requestId: "d" });
|
||||
queryClient.setQueryData(terminalKey, { cwd: "/repo", terminals: [], requestId: "t" });
|
||||
queryClient.setQueryData(otherProviderKey, {
|
||||
entries: [],
|
||||
generatedAt: "now",
|
||||
requestId: "other",
|
||||
});
|
||||
|
||||
invalidateServerDataQueriesAfterReconnect({ queryClient, serverId });
|
||||
|
||||
expect(queryClient.getQueryState(providerKey)?.isInvalidated).toBe(true);
|
||||
expect(queryClient.getQueryState(daemonConfigKey)?.isInvalidated).toBe(true);
|
||||
expect(queryClient.getQueryState(diffKey)?.isInvalidated).toBe(true);
|
||||
expect(queryClient.getQueryState(terminalKey)?.isInvalidated).toBe(true);
|
||||
expect(queryClient.getQueryState(otherProviderKey)?.isInvalidated).toBe(false);
|
||||
});
|
||||
});
|
||||
@@ -0,0 +1,754 @@
|
||||
import type { Query, QueryCacheNotifyEvent, QueryClient, QueryKey } from "@tanstack/react-query";
|
||||
import type {
|
||||
ListTerminalsResponse,
|
||||
MutableDaemonConfig,
|
||||
SessionOutboundMessage,
|
||||
} from "@getpaseo/protocol/messages";
|
||||
import { agentCommandsQueryRoot } from "@/hooks/agent-commands-query";
|
||||
import { orderCheckoutDiffFiles } from "@/git/diff-order";
|
||||
import { daemonConfigQueryKey } from "@/data/daemon-config";
|
||||
import { providersSnapshotQueryKey, providersSnapshotQueryRoot } from "@/data/providers-snapshot";
|
||||
|
||||
type ProvidersSnapshotUpdateMessage = Extract<
|
||||
SessionOutboundMessage,
|
||||
{ type: "providers_snapshot_update" }
|
||||
>;
|
||||
type CheckoutDiffUpdateMessage = Extract<SessionOutboundMessage, { type: "checkout_diff_update" }>;
|
||||
type SubscribeCheckoutDiffResponseMessage = Extract<
|
||||
SessionOutboundMessage,
|
||||
{ type: "subscribe_checkout_diff_response" }
|
||||
>;
|
||||
type StatusMessage = Extract<SessionOutboundMessage, { type: "status" }>;
|
||||
type TerminalsChangedMessage = Extract<SessionOutboundMessage, { type: "terminals_changed" }>;
|
||||
type ServerDataEventType =
|
||||
| "providers_snapshot_update"
|
||||
| "checkout_diff_update"
|
||||
| "subscribe_checkout_diff_response"
|
||||
| "status"
|
||||
| "terminals_changed";
|
||||
type CheckoutDiffResponsePayload = SubscribeCheckoutDiffResponseMessage["payload"];
|
||||
type CheckoutDiffCachePayload = Omit<CheckoutDiffResponsePayload, "subscriptionId">;
|
||||
type ListTerminalsPayload = ListTerminalsResponse["payload"];
|
||||
|
||||
interface CheckoutDiffCompare {
|
||||
mode: "uncommitted" | "base";
|
||||
baseRef?: string;
|
||||
ignoreWhitespace?: boolean;
|
||||
}
|
||||
|
||||
interface CheckoutDiffRoute {
|
||||
domain: "checkoutDiff";
|
||||
enabled: boolean;
|
||||
serverId: string;
|
||||
subscriptionId: string;
|
||||
cwd: string;
|
||||
compare: CheckoutDiffCompare;
|
||||
}
|
||||
|
||||
interface WorkspaceTerminalsRoute {
|
||||
domain: "workspaceTerminals";
|
||||
enabled: boolean;
|
||||
serverId: string;
|
||||
cwd: string;
|
||||
workspaceId?: string;
|
||||
}
|
||||
|
||||
type ServerDataRoute = CheckoutDiffRoute | WorkspaceTerminalsRoute;
|
||||
|
||||
export interface ServerDataQueryMeta extends Record<string, unknown> {
|
||||
serverData: ServerDataRoute;
|
||||
}
|
||||
|
||||
export type ProvidersSnapshotUpdate = ProvidersSnapshotUpdateMessage;
|
||||
|
||||
interface ServerDataPushClient {
|
||||
on<TType extends ServerDataEventType>(
|
||||
type: TType,
|
||||
handler: (message: Extract<SessionOutboundMessage, { type: TType }>) => void,
|
||||
): () => void;
|
||||
subscribeCheckoutDiff(
|
||||
cwd: string,
|
||||
compare: CheckoutDiffCompare,
|
||||
options: { subscriptionId: string; requestId?: string },
|
||||
): Promise<CheckoutDiffResponsePayload>;
|
||||
unsubscribeCheckoutDiff(subscriptionId: string): void;
|
||||
subscribeTerminals(input: { cwd: string; workspaceId?: string }): void;
|
||||
unsubscribeTerminals(input: { cwd: string; workspaceId?: string }): void;
|
||||
}
|
||||
|
||||
interface PushRouterInput {
|
||||
client: ServerDataPushClient;
|
||||
queryClient: QueryClient;
|
||||
serverId: string;
|
||||
}
|
||||
|
||||
interface ActiveServerDataSubscriptions {
|
||||
checkoutDiff: Map<string, CheckoutDiffRoute>;
|
||||
workspaceTerminals: Map<string, WorkspaceTerminalsRoute>;
|
||||
}
|
||||
|
||||
interface ReconnectRepairPolicy {
|
||||
domain: string;
|
||||
invalidate(input: { queryClient: QueryClient; serverId: string }): void;
|
||||
}
|
||||
|
||||
const RECONNECT_REPAIR_POLICIES: ReconnectRepairPolicy[] = [
|
||||
{
|
||||
domain: "providersSnapshot",
|
||||
invalidate: ({ queryClient, serverId }) => {
|
||||
void queryClient.invalidateQueries({ queryKey: providersSnapshotQueryRoot(serverId) });
|
||||
},
|
||||
},
|
||||
{
|
||||
domain: "daemonConfig",
|
||||
invalidate: ({ queryClient, serverId }) => {
|
||||
void queryClient.invalidateQueries({ queryKey: daemonConfigQueryKey(serverId) });
|
||||
},
|
||||
},
|
||||
{
|
||||
domain: "checkoutDiff",
|
||||
invalidate: ({ queryClient, serverId }) => {
|
||||
void queryClient.invalidateQueries({
|
||||
predicate: (query) => isQueryForServer(query.queryKey, "checkoutDiff", serverId),
|
||||
});
|
||||
},
|
||||
},
|
||||
{
|
||||
domain: "workspaceTerminals",
|
||||
invalidate: ({ queryClient, serverId }) => {
|
||||
void queryClient.invalidateQueries({
|
||||
predicate: (query) => isQueryForServer(query.queryKey, "terminals", serverId),
|
||||
});
|
||||
},
|
||||
},
|
||||
];
|
||||
const reconnectSubscriptionRepairsByServerId = new Map<string, Set<() => void>>();
|
||||
|
||||
export function checkoutDiffPushRoute(input: {
|
||||
enabled: boolean;
|
||||
serverId: string;
|
||||
subscriptionId: string;
|
||||
cwd: string;
|
||||
compare: CheckoutDiffCompare;
|
||||
}): ServerDataQueryMeta {
|
||||
return {
|
||||
serverData: {
|
||||
domain: "checkoutDiff",
|
||||
enabled: input.enabled,
|
||||
serverId: input.serverId,
|
||||
subscriptionId: input.subscriptionId,
|
||||
cwd: input.cwd,
|
||||
compare: input.compare,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function workspaceTerminalsPushRoute(input: {
|
||||
enabled: boolean;
|
||||
serverId: string;
|
||||
cwd: string;
|
||||
workspaceId?: string;
|
||||
}): ServerDataQueryMeta {
|
||||
return {
|
||||
serverData: {
|
||||
domain: "workspaceTerminals",
|
||||
enabled: input.enabled,
|
||||
serverId: input.serverId,
|
||||
cwd: input.cwd,
|
||||
...(input.workspaceId ? { workspaceId: input.workspaceId } : {}),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
export function invalidateServerDataQueriesAfterReconnect(input: {
|
||||
queryClient: QueryClient;
|
||||
serverId: string;
|
||||
}): void {
|
||||
for (const policy of RECONNECT_REPAIR_POLICIES) {
|
||||
policy.invalidate(input);
|
||||
}
|
||||
for (const repairSubscriptions of reconnectSubscriptionRepairsByServerId.get(input.serverId) ??
|
||||
[]) {
|
||||
repairSubscriptions();
|
||||
}
|
||||
}
|
||||
|
||||
export function applyProvidersSnapshotUpdate(input: {
|
||||
serverId: string;
|
||||
queryClient: QueryClient;
|
||||
message: ProvidersSnapshotUpdate;
|
||||
}): void {
|
||||
if (input.message.type !== "providers_snapshot_update") {
|
||||
return;
|
||||
}
|
||||
const queryKey = providersSnapshotQueryKey(input.serverId, input.message.payload.cwd);
|
||||
input.queryClient.setQueryData(queryKey, {
|
||||
entries: input.message.payload.entries,
|
||||
generatedAt: input.message.payload.generatedAt,
|
||||
requestId: "providers_snapshot_update",
|
||||
});
|
||||
void input.queryClient.invalidateQueries({
|
||||
queryKey: agentCommandsQueryRoot(input.serverId),
|
||||
exact: false,
|
||||
});
|
||||
}
|
||||
|
||||
export function mountServerDataPushRouter(input: PushRouterInput): () => void {
|
||||
const activeCheckoutDiffSubscriptions = new Map<string, CheckoutDiffRoute>();
|
||||
const activeTerminalSubscriptions = new Map<string, WorkspaceTerminalsRoute>();
|
||||
let disposed = false;
|
||||
|
||||
function reconcileSubscriptions(
|
||||
fallbackActive: ActiveServerDataSubscriptions = {
|
||||
checkoutDiff: activeCheckoutDiffSubscriptions,
|
||||
workspaceTerminals: activeTerminalSubscriptions,
|
||||
},
|
||||
): void {
|
||||
if (disposed) {
|
||||
return;
|
||||
}
|
||||
|
||||
const desiredCheckoutDiffSubscriptions = new Map<string, CheckoutDiffRoute>();
|
||||
const desiredTerminalSubscriptions = new Map<string, WorkspaceTerminalsRoute>();
|
||||
for (const query of input.queryClient.getQueryCache().getAll()) {
|
||||
const route = getActiveServerDataRoute(query, input.serverId, {
|
||||
checkoutDiff: fallbackActive.checkoutDiff,
|
||||
workspaceTerminals: fallbackActive.workspaceTerminals,
|
||||
});
|
||||
if (!route) {
|
||||
continue;
|
||||
}
|
||||
if (route.domain === "checkoutDiff") {
|
||||
desiredCheckoutDiffSubscriptions.set(route.subscriptionId, route);
|
||||
continue;
|
||||
}
|
||||
desiredTerminalSubscriptions.set(workspaceTerminalSubscriptionKey(route), route);
|
||||
}
|
||||
|
||||
reconcileCheckoutDiffSubscriptions({
|
||||
active: activeCheckoutDiffSubscriptions,
|
||||
client: input.client,
|
||||
desired: desiredCheckoutDiffSubscriptions,
|
||||
serverId: input.serverId,
|
||||
});
|
||||
reconcileTerminalSubscriptions({
|
||||
active: activeTerminalSubscriptions,
|
||||
client: input.client,
|
||||
desired: desiredTerminalSubscriptions,
|
||||
});
|
||||
}
|
||||
|
||||
function resetSubscriptionsAfterReconnect(): void {
|
||||
const fallbackActive = {
|
||||
checkoutDiff: new Map(activeCheckoutDiffSubscriptions),
|
||||
workspaceTerminals: new Map(activeTerminalSubscriptions),
|
||||
};
|
||||
activeCheckoutDiffSubscriptions.clear();
|
||||
activeTerminalSubscriptions.clear();
|
||||
reconcileSubscriptions(fallbackActive);
|
||||
}
|
||||
|
||||
const unsubscribeQueryCache = input.queryClient.getQueryCache().subscribe((event) => {
|
||||
if (
|
||||
!shouldReconcileSubscriptionsForCacheEvent(event, input.serverId, {
|
||||
checkoutDiff: activeCheckoutDiffSubscriptions,
|
||||
workspaceTerminals: activeTerminalSubscriptions,
|
||||
})
|
||||
) {
|
||||
return;
|
||||
}
|
||||
reconcileSubscriptions();
|
||||
});
|
||||
const unsubscribeProviders = input.client.on("providers_snapshot_update", (message) => {
|
||||
applyProvidersSnapshotUpdate({
|
||||
queryClient: input.queryClient,
|
||||
serverId: input.serverId,
|
||||
message,
|
||||
});
|
||||
});
|
||||
const unsubscribeDaemonConfig = input.client.on("status", (message) => {
|
||||
applyDaemonConfigStatus({ queryClient: input.queryClient, serverId: input.serverId, message });
|
||||
});
|
||||
const unsubscribeCheckoutDiffUpdate = input.client.on("checkout_diff_update", (message) => {
|
||||
applyCheckoutDiffUpdate({
|
||||
activeCheckoutDiffSubscriptions,
|
||||
queryClient: input.queryClient,
|
||||
serverId: input.serverId,
|
||||
message,
|
||||
});
|
||||
});
|
||||
const unsubscribeCheckoutDiffResponse = input.client.on(
|
||||
"subscribe_checkout_diff_response",
|
||||
(message) => {
|
||||
applyCheckoutDiffSubscribeResponse({
|
||||
activeCheckoutDiffSubscriptions,
|
||||
queryClient: input.queryClient,
|
||||
serverId: input.serverId,
|
||||
message,
|
||||
});
|
||||
},
|
||||
);
|
||||
const unsubscribeTerminalsChanged = input.client.on("terminals_changed", (message) => {
|
||||
applyTerminalsChanged({
|
||||
activeCheckoutDiffSubscriptions,
|
||||
activeTerminalSubscriptions,
|
||||
queryClient: input.queryClient,
|
||||
serverId: input.serverId,
|
||||
message,
|
||||
});
|
||||
});
|
||||
let reconnectSubscriptionRepairs = reconnectSubscriptionRepairsByServerId.get(input.serverId);
|
||||
if (!reconnectSubscriptionRepairs) {
|
||||
reconnectSubscriptionRepairs = new Set();
|
||||
reconnectSubscriptionRepairsByServerId.set(input.serverId, reconnectSubscriptionRepairs);
|
||||
}
|
||||
reconnectSubscriptionRepairs.add(resetSubscriptionsAfterReconnect);
|
||||
|
||||
reconcileSubscriptions();
|
||||
|
||||
return () => {
|
||||
disposed = true;
|
||||
reconnectSubscriptionRepairs.delete(resetSubscriptionsAfterReconnect);
|
||||
if (reconnectSubscriptionRepairs.size === 0) {
|
||||
reconnectSubscriptionRepairsByServerId.delete(input.serverId);
|
||||
}
|
||||
unsubscribeQueryCache();
|
||||
unsubscribeProviders();
|
||||
unsubscribeDaemonConfig();
|
||||
unsubscribeCheckoutDiffUpdate();
|
||||
unsubscribeCheckoutDiffResponse();
|
||||
unsubscribeTerminalsChanged();
|
||||
for (const subscriptionId of activeCheckoutDiffSubscriptions.keys()) {
|
||||
unsubscribeCheckoutDiff(input.client, subscriptionId);
|
||||
}
|
||||
activeCheckoutDiffSubscriptions.clear();
|
||||
for (const route of activeTerminalSubscriptions.values()) {
|
||||
input.client.unsubscribeTerminals(workspaceTerminalSubscriptionInput(route));
|
||||
}
|
||||
activeTerminalSubscriptions.clear();
|
||||
};
|
||||
}
|
||||
|
||||
function reconcileCheckoutDiffSubscriptions(input: {
|
||||
active: Map<string, CheckoutDiffRoute>;
|
||||
client: ServerDataPushClient;
|
||||
desired: Map<string, CheckoutDiffRoute>;
|
||||
serverId: string;
|
||||
}): void {
|
||||
for (const [subscriptionId, current] of input.active) {
|
||||
const desired = input.desired.get(subscriptionId);
|
||||
if (desired && areCheckoutDiffRoutesEqual(current, desired)) {
|
||||
continue;
|
||||
}
|
||||
unsubscribeCheckoutDiff(input.client, subscriptionId);
|
||||
input.active.delete(subscriptionId);
|
||||
}
|
||||
|
||||
for (const [subscriptionId, desired] of input.desired) {
|
||||
if (input.active.has(subscriptionId)) {
|
||||
continue;
|
||||
}
|
||||
input.active.set(subscriptionId, desired);
|
||||
void input.client
|
||||
.subscribeCheckoutDiff(desired.cwd, desired.compare, {
|
||||
subscriptionId,
|
||||
requestId: `push-router:${input.serverId}:${subscriptionId}`,
|
||||
})
|
||||
.catch((error) => {
|
||||
if (areCheckoutDiffRoutesEqual(input.active.get(subscriptionId), desired)) {
|
||||
input.active.delete(subscriptionId);
|
||||
}
|
||||
console.error("[server-data] subscribeCheckoutDiff failed", {
|
||||
serverId: input.serverId,
|
||||
cwd: desired.cwd,
|
||||
error,
|
||||
});
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
function reconcileTerminalSubscriptions(input: {
|
||||
active: Map<string, WorkspaceTerminalsRoute>;
|
||||
client: ServerDataPushClient;
|
||||
desired: Map<string, WorkspaceTerminalsRoute>;
|
||||
}): void {
|
||||
for (const [key, current] of input.active) {
|
||||
const desired = input.desired.get(key);
|
||||
if (desired && areWorkspaceTerminalsRoutesEqual(current, desired)) {
|
||||
continue;
|
||||
}
|
||||
input.client.unsubscribeTerminals(workspaceTerminalSubscriptionInput(current));
|
||||
input.active.delete(key);
|
||||
}
|
||||
|
||||
for (const [key, desired] of input.desired) {
|
||||
if (input.active.has(key)) {
|
||||
continue;
|
||||
}
|
||||
input.active.set(key, desired);
|
||||
input.client.subscribeTerminals(workspaceTerminalSubscriptionInput(desired));
|
||||
}
|
||||
}
|
||||
|
||||
function applyDaemonConfigStatus(input: {
|
||||
queryClient: QueryClient;
|
||||
serverId: string;
|
||||
message: StatusMessage;
|
||||
}): void {
|
||||
const payload = input.message.payload;
|
||||
if (!isDaemonConfigChangedPayload(payload)) {
|
||||
return;
|
||||
}
|
||||
input.queryClient.setQueryData<MutableDaemonConfig>(
|
||||
daemonConfigQueryKey(input.serverId),
|
||||
payload.config,
|
||||
);
|
||||
}
|
||||
|
||||
function applyCheckoutDiffUpdate(input: {
|
||||
activeCheckoutDiffSubscriptions: Map<string, CheckoutDiffRoute>;
|
||||
queryClient: QueryClient;
|
||||
serverId: string;
|
||||
message: CheckoutDiffUpdateMessage;
|
||||
}): void {
|
||||
setCheckoutDiffPayload({
|
||||
activeCheckoutDiffSubscriptions: input.activeCheckoutDiffSubscriptions,
|
||||
queryClient: input.queryClient,
|
||||
serverId: input.serverId,
|
||||
subscriptionId: input.message.payload.subscriptionId,
|
||||
payload: {
|
||||
cwd: input.message.payload.cwd,
|
||||
files: orderCheckoutDiffFiles(input.message.payload.files),
|
||||
error: input.message.payload.error,
|
||||
requestId: `subscription:${input.message.payload.subscriptionId}`,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function applyCheckoutDiffSubscribeResponse(input: {
|
||||
activeCheckoutDiffSubscriptions: Map<string, CheckoutDiffRoute>;
|
||||
queryClient: QueryClient;
|
||||
serverId: string;
|
||||
message: SubscribeCheckoutDiffResponseMessage;
|
||||
}): void {
|
||||
setCheckoutDiffPayload({
|
||||
activeCheckoutDiffSubscriptions: input.activeCheckoutDiffSubscriptions,
|
||||
queryClient: input.queryClient,
|
||||
serverId: input.serverId,
|
||||
subscriptionId: input.message.payload.subscriptionId,
|
||||
payload: {
|
||||
cwd: input.message.payload.cwd,
|
||||
files: orderCheckoutDiffFiles(input.message.payload.files),
|
||||
error: input.message.payload.error,
|
||||
requestId: input.message.payload.requestId,
|
||||
},
|
||||
});
|
||||
}
|
||||
|
||||
function setCheckoutDiffPayload(input: {
|
||||
activeCheckoutDiffSubscriptions: Map<string, CheckoutDiffRoute>;
|
||||
queryClient: QueryClient;
|
||||
serverId: string;
|
||||
subscriptionId: string;
|
||||
payload: CheckoutDiffCachePayload;
|
||||
}): void {
|
||||
for (const query of input.queryClient.getQueryCache().getAll()) {
|
||||
const route =
|
||||
getServerDataRoute(query) ??
|
||||
getActiveCheckoutDiffRouteForQueryKey({
|
||||
active: input.activeCheckoutDiffSubscriptions,
|
||||
queryKey: query.queryKey,
|
||||
serverId: input.serverId,
|
||||
});
|
||||
if (
|
||||
!route ||
|
||||
route.domain !== "checkoutDiff" ||
|
||||
route.serverId !== input.serverId ||
|
||||
route.subscriptionId !== input.subscriptionId
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
input.queryClient.setQueryData<CheckoutDiffCachePayload>(query.queryKey, input.payload);
|
||||
}
|
||||
}
|
||||
|
||||
function applyTerminalsChanged(input: {
|
||||
activeCheckoutDiffSubscriptions: Map<string, CheckoutDiffRoute>;
|
||||
activeTerminalSubscriptions: Map<string, WorkspaceTerminalsRoute>;
|
||||
queryClient: QueryClient;
|
||||
serverId: string;
|
||||
message: TerminalsChangedMessage;
|
||||
}): void {
|
||||
for (const query of input.queryClient.getQueryCache().getAll()) {
|
||||
const route = getActiveServerDataRoute(query, input.serverId, {
|
||||
checkoutDiff: input.activeCheckoutDiffSubscriptions,
|
||||
workspaceTerminals: input.activeTerminalSubscriptions,
|
||||
});
|
||||
if (
|
||||
!route ||
|
||||
route.domain !== "workspaceTerminals" ||
|
||||
route.cwd !== input.message.payload.cwd
|
||||
) {
|
||||
continue;
|
||||
}
|
||||
|
||||
const matchingTerminals = input.message.payload.terminals.filter(
|
||||
(terminal) => terminal.workspaceId === route.workspaceId,
|
||||
);
|
||||
|
||||
input.queryClient.setQueryData<ListTerminalsPayload>(query.queryKey, (current) => ({
|
||||
cwd: input.message.payload.cwd,
|
||||
terminals: matchingTerminals,
|
||||
requestId: current?.requestId ?? `terminals-changed-${Date.now()}`,
|
||||
}));
|
||||
}
|
||||
}
|
||||
|
||||
function getActiveServerDataRoute(
|
||||
query: Query,
|
||||
serverId: string,
|
||||
active: ActiveServerDataSubscriptions,
|
||||
): ServerDataRoute | null {
|
||||
if (query.getObserversCount() === 0) {
|
||||
return null;
|
||||
}
|
||||
const route = getServerDataRoute(query);
|
||||
if (route) {
|
||||
return route.enabled && route.serverId === serverId ? route : null;
|
||||
}
|
||||
return getActiveRouteForQueryKey({
|
||||
active,
|
||||
queryKey: query.queryKey,
|
||||
serverId,
|
||||
});
|
||||
}
|
||||
|
||||
function getActiveRouteForQueryKey(input: {
|
||||
active: ActiveServerDataSubscriptions;
|
||||
queryKey: QueryKey;
|
||||
serverId: string;
|
||||
}): ServerDataRoute | null {
|
||||
return (
|
||||
getActiveTerminalRouteForQueryKey({
|
||||
active: input.active.workspaceTerminals,
|
||||
queryKey: input.queryKey,
|
||||
serverId: input.serverId,
|
||||
}) ??
|
||||
getActiveCheckoutDiffRouteForQueryKey({
|
||||
active: input.active.checkoutDiff,
|
||||
queryKey: input.queryKey,
|
||||
serverId: input.serverId,
|
||||
})
|
||||
);
|
||||
}
|
||||
|
||||
function getActiveTerminalRouteForQueryKey(input: {
|
||||
active: Map<string, WorkspaceTerminalsRoute>;
|
||||
queryKey: QueryKey;
|
||||
serverId: string;
|
||||
}): WorkspaceTerminalsRoute | null {
|
||||
if (!isQueryForServer(input.queryKey, "terminals", input.serverId)) {
|
||||
return null;
|
||||
}
|
||||
const cwd = input.queryKey[2];
|
||||
const workspaceId = input.queryKey[3];
|
||||
if (
|
||||
typeof cwd !== "string" ||
|
||||
(workspaceId !== undefined && workspaceId !== null && typeof workspaceId !== "string")
|
||||
) {
|
||||
return null;
|
||||
}
|
||||
return input.active.get(`${cwd}\u0000${workspaceId ?? ""}`) ?? null;
|
||||
}
|
||||
|
||||
function getActiveCheckoutDiffRouteForQueryKey(input: {
|
||||
active: Map<string, CheckoutDiffRoute>;
|
||||
queryKey: QueryKey;
|
||||
serverId: string;
|
||||
}): CheckoutDiffRoute | null {
|
||||
if (!isQueryForServer(input.queryKey, "checkoutDiff", input.serverId)) {
|
||||
return null;
|
||||
}
|
||||
for (const route of input.active.values()) {
|
||||
if (isCheckoutDiffQueryKeyForRoute(input.queryKey, route)) {
|
||||
return route;
|
||||
}
|
||||
}
|
||||
return null;
|
||||
}
|
||||
|
||||
function shouldReconcileSubscriptionsForCacheEvent(
|
||||
event: QueryCacheNotifyEvent,
|
||||
serverId: string,
|
||||
active: ActiveServerDataSubscriptions,
|
||||
): boolean {
|
||||
if (!canEventChangeDesiredSubscriptions(event.type)) {
|
||||
return false;
|
||||
}
|
||||
const route = getServerDataRoute(event.query);
|
||||
if (route?.serverId === serverId) {
|
||||
return true;
|
||||
}
|
||||
return (
|
||||
getActiveRouteForQueryKey({
|
||||
active,
|
||||
queryKey: event.query.queryKey,
|
||||
serverId,
|
||||
}) !== null
|
||||
);
|
||||
}
|
||||
|
||||
function canEventChangeDesiredSubscriptions(type: QueryCacheNotifyEvent["type"]): boolean {
|
||||
return (
|
||||
type === "added" ||
|
||||
type === "removed" ||
|
||||
type === "observerAdded" ||
|
||||
type === "observerRemoved" ||
|
||||
type === "observerOptionsUpdated"
|
||||
);
|
||||
}
|
||||
|
||||
function getServerDataRoute(query: Query): ServerDataRoute | null {
|
||||
const meta = query.meta;
|
||||
if (!isRecord(meta) || !isRecord(meta.serverData)) {
|
||||
return null;
|
||||
}
|
||||
return readServerDataRoute(meta.serverData);
|
||||
}
|
||||
|
||||
function readServerDataRoute(value: Record<string, unknown>): ServerDataRoute | null {
|
||||
const domain = value.domain;
|
||||
const enabled = value.enabled;
|
||||
const serverId = value.serverId;
|
||||
const cwd = value.cwd;
|
||||
if (typeof enabled !== "boolean" || typeof serverId !== "string" || typeof cwd !== "string") {
|
||||
return null;
|
||||
}
|
||||
|
||||
if (domain === "checkoutDiff") {
|
||||
const subscriptionId = value.subscriptionId;
|
||||
const compare = readCheckoutDiffCompare(value.compare);
|
||||
if (typeof subscriptionId !== "string" || !compare) {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
domain,
|
||||
enabled,
|
||||
serverId,
|
||||
subscriptionId,
|
||||
cwd,
|
||||
compare,
|
||||
};
|
||||
}
|
||||
|
||||
if (domain === "workspaceTerminals") {
|
||||
const workspaceId = value.workspaceId;
|
||||
if (workspaceId !== undefined && typeof workspaceId !== "string") {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
domain,
|
||||
enabled,
|
||||
serverId,
|
||||
cwd,
|
||||
...(workspaceId ? { workspaceId } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
return null;
|
||||
}
|
||||
|
||||
function readCheckoutDiffCompare(value: unknown): CheckoutDiffCompare | null {
|
||||
if (!isRecord(value)) {
|
||||
return null;
|
||||
}
|
||||
const mode = value.mode;
|
||||
const baseRef = value.baseRef;
|
||||
const ignoreWhitespace = value.ignoreWhitespace;
|
||||
if (mode !== "uncommitted" && mode !== "base") {
|
||||
return null;
|
||||
}
|
||||
if (baseRef !== undefined && typeof baseRef !== "string") {
|
||||
return null;
|
||||
}
|
||||
if (ignoreWhitespace !== undefined && typeof ignoreWhitespace !== "boolean") {
|
||||
return null;
|
||||
}
|
||||
return {
|
||||
mode,
|
||||
...(baseRef ? { baseRef } : {}),
|
||||
...(ignoreWhitespace !== undefined ? { ignoreWhitespace } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
function areCheckoutDiffRoutesEqual(
|
||||
left: CheckoutDiffRoute | undefined,
|
||||
right: CheckoutDiffRoute,
|
||||
): boolean {
|
||||
return (
|
||||
left?.serverId === right.serverId &&
|
||||
left.subscriptionId === right.subscriptionId &&
|
||||
left.cwd === right.cwd &&
|
||||
left.compare.mode === right.compare.mode &&
|
||||
left.compare.baseRef === right.compare.baseRef &&
|
||||
left.compare.ignoreWhitespace === right.compare.ignoreWhitespace
|
||||
);
|
||||
}
|
||||
|
||||
function isCheckoutDiffQueryKeyForRoute(queryKey: QueryKey, route: CheckoutDiffRoute): boolean {
|
||||
return (
|
||||
queryKey[0] === "checkoutDiff" &&
|
||||
queryKey[1] === route.serverId &&
|
||||
queryKey[2] === route.cwd &&
|
||||
queryKey[3] === route.compare.mode &&
|
||||
queryKey[4] === (route.compare.baseRef ?? "") &&
|
||||
queryKey[5] === (route.compare.ignoreWhitespace === true)
|
||||
);
|
||||
}
|
||||
|
||||
function areWorkspaceTerminalsRoutesEqual(
|
||||
left: WorkspaceTerminalsRoute,
|
||||
right: WorkspaceTerminalsRoute,
|
||||
): boolean {
|
||||
return (
|
||||
left.serverId === right.serverId &&
|
||||
left.cwd === right.cwd &&
|
||||
left.workspaceId === right.workspaceId
|
||||
);
|
||||
}
|
||||
|
||||
function workspaceTerminalSubscriptionKey(route: WorkspaceTerminalsRoute): string {
|
||||
return `${route.cwd}\u0000${route.workspaceId ?? ""}`;
|
||||
}
|
||||
|
||||
function workspaceTerminalSubscriptionInput(route: WorkspaceTerminalsRoute): {
|
||||
cwd: string;
|
||||
workspaceId?: string;
|
||||
} {
|
||||
return {
|
||||
cwd: route.cwd,
|
||||
...(route.workspaceId ? { workspaceId: route.workspaceId } : {}),
|
||||
};
|
||||
}
|
||||
|
||||
function unsubscribeCheckoutDiff(client: ServerDataPushClient, subscriptionId: string): void {
|
||||
try {
|
||||
client.unsubscribeCheckoutDiff(subscriptionId);
|
||||
} catch {
|
||||
// Disconnect cleanup can race with explicit subscription teardown.
|
||||
}
|
||||
}
|
||||
|
||||
function isQueryForServer(queryKey: QueryKey, kind: string, serverId: string): boolean {
|
||||
return queryKey.length >= 2 && queryKey[0] === kind && queryKey[1] === serverId;
|
||||
}
|
||||
|
||||
function isRecord(value: unknown): value is Record<string, unknown> {
|
||||
return typeof value === "object" && value !== null;
|
||||
}
|
||||
|
||||
function isDaemonConfigChangedPayload(
|
||||
payload: StatusMessage["payload"],
|
||||
): payload is { status: "daemon_config_changed"; config: MutableDaemonConfig } {
|
||||
return payload.status === "daemon_config_changed" && isRecord(payload.config);
|
||||
}
|
||||
@@ -0,0 +1,13 @@
|
||||
import { QueryClient } from "@tanstack/react-query";
|
||||
|
||||
export const queryClient = new QueryClient({
|
||||
defaultOptions: {
|
||||
queries: {
|
||||
staleTime: Infinity,
|
||||
gcTime: Infinity,
|
||||
refetchOnMount: false,
|
||||
refetchOnReconnect: false,
|
||||
refetchOnWindowFocus: false,
|
||||
},
|
||||
},
|
||||
});
|
||||
@@ -0,0 +1,106 @@
|
||||
import {
|
||||
keepPreviousData,
|
||||
skipToken,
|
||||
useQuery,
|
||||
type QueryKey,
|
||||
type UseQueryOptions,
|
||||
type UseQueryResult,
|
||||
} from "@tanstack/react-query";
|
||||
|
||||
type QueryFnOption<TQueryFnData, TError, TData, TQueryKey extends QueryKey> = NonNullable<
|
||||
UseQueryOptions<TQueryFnData, TError, TData, TQueryKey>["queryFn"]
|
||||
>;
|
||||
|
||||
type ReplicaQueryInput<TQueryFnData, TError, TData, TQueryKey extends QueryKey> = Omit<
|
||||
UseQueryOptions<TQueryFnData, TError, TData, TQueryKey>,
|
||||
| "gcTime"
|
||||
| "initialData"
|
||||
| "refetchOnMount"
|
||||
| "refetchOnReconnect"
|
||||
| "refetchOnWindowFocus"
|
||||
| "staleTime"
|
||||
> & {
|
||||
pushEvent: string;
|
||||
};
|
||||
|
||||
type FetchQueryInput<TQueryFnData, TError, TData, TQueryKey extends QueryKey> = Omit<
|
||||
UseQueryOptions<TQueryFnData, TError, TData, TQueryKey>,
|
||||
"initialData" | "placeholderData" | "queryFn" | "refetchOnMount" | "staleTime"
|
||||
> & {
|
||||
dataShape: "list" | "value";
|
||||
queryFn: QueryFnOption<TQueryFnData, TError, TData, TQueryKey>;
|
||||
staleTimeMs: number;
|
||||
};
|
||||
|
||||
export function useReplicaQuery<
|
||||
TQueryFnData,
|
||||
TError = Error,
|
||||
TData = TQueryFnData,
|
||||
TQueryKey extends QueryKey = QueryKey,
|
||||
>(input: ReplicaQueryInput<TQueryFnData, TError, TData, TQueryKey>): UseQueryResult<TData, TError> {
|
||||
return useQuery(replicaQueryOptions(input));
|
||||
}
|
||||
|
||||
export function useFetchQuery<
|
||||
TQueryFnData,
|
||||
TError = Error,
|
||||
TData = TQueryFnData,
|
||||
TQueryKey extends QueryKey = QueryKey,
|
||||
>(input: FetchQueryInput<TQueryFnData, TError, TData, TQueryKey>): UseQueryResult<TData, TError> {
|
||||
return useQuery(fetchQueryOptions(input));
|
||||
}
|
||||
|
||||
function replicaQueryOptions<
|
||||
TQueryFnData,
|
||||
TError = Error,
|
||||
TData = TQueryFnData,
|
||||
TQueryKey extends QueryKey = QueryKey,
|
||||
>(
|
||||
input: ReplicaQueryInput<TQueryFnData, TError, TData, TQueryKey>,
|
||||
): UseQueryOptions<TQueryFnData, TError, TData, TQueryKey> {
|
||||
const { pushEvent, meta, ...options } = input;
|
||||
return {
|
||||
...options,
|
||||
gcTime: Infinity,
|
||||
meta: {
|
||||
...meta,
|
||||
serverDataPolicy: {
|
||||
class: "replica",
|
||||
pushEvent,
|
||||
},
|
||||
},
|
||||
queryFn: options.queryFn ?? skipToken,
|
||||
refetchOnMount: false,
|
||||
refetchOnReconnect: false,
|
||||
refetchOnWindowFocus: false,
|
||||
staleTime: Infinity,
|
||||
};
|
||||
}
|
||||
|
||||
function fetchQueryOptions<
|
||||
TQueryFnData,
|
||||
TError = Error,
|
||||
TData = TQueryFnData,
|
||||
TQueryKey extends QueryKey = QueryKey,
|
||||
>(
|
||||
input: FetchQueryInput<TQueryFnData, TError, TData, TQueryKey>,
|
||||
): UseQueryOptions<TQueryFnData, TError, TData, TQueryKey> {
|
||||
if (!Number.isFinite(input.staleTimeMs)) {
|
||||
throw new Error("Fetch queries must declare a finite staleTimeMs.");
|
||||
}
|
||||
|
||||
const { dataShape, meta, staleTimeMs, ...options } = input;
|
||||
return {
|
||||
...options,
|
||||
...(dataShape === "list" ? { placeholderData: keepPreviousData } : {}),
|
||||
meta: {
|
||||
...meta,
|
||||
serverDataPolicy: {
|
||||
class: "fetch",
|
||||
dataShape,
|
||||
},
|
||||
},
|
||||
refetchOnMount: "always",
|
||||
staleTime: staleTimeMs,
|
||||
};
|
||||
}
|
||||
Reference in New Issue
Block a user