Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 10 additions & 4 deletions packages/cli/src/server/studioServer.ts
Original file line number Diff line number Diff line change
Expand Up @@ -790,11 +790,17 @@ export function createStudioServer(options: StudioServerOptions): StudioServer {
};
// Re-applied here because the watcher now also emits the signature
// manifest files, which must not trigger a browser reload.
watcher.addListener((changedPath) => {
const wrappedListener = (changedPath: string) => {
if (shouldWatchProjectFile(changedPath)) listener(changedPath);
});
while (true) {
await stream.sleep(30000);
};
watcher.addListener(wrappedListener);
stream.onAbort(() => watcher.removeListener(wrappedListener));
try {
while (true) {
await stream.sleep(30000);
}
} finally {
watcher.removeListener(wrappedListener);
}
});
});
Expand Down
Original file line number Diff line number Diff line change
Expand Up @@ -175,7 +175,7 @@ describe("external file change coordinator", () => {
expect(options.reloadSdkSession).toHaveBeenCalledOnce();
});

it("ignores stale drain completion after a newer generation", async () => {
it("serializes drains and processes stashed events", async () => {
const drains: Array<(result: { status: "clean" }) => void> = [];
const { options } = await mountCoordinator({
drainPendingChanges: () => new Promise((resolve) => drains.push(resolve)),
Expand All @@ -184,11 +184,14 @@ describe("external file change coordinator", () => {
handler?.({ path: "index.html", content: "first", version: "v2" });
handler?.({ path: "index.html", content: "second", version: "v3" });
});
expect(drains).toHaveLength(1);
await act(async () => drains[0]?.({ status: "clean" }));
expect(options.reloadPreview).not.toHaveBeenCalled();
await act(async () => drains[1]?.({ status: "clean" }));
expect(options.reloadPreview).toHaveBeenCalledOnce();
expect(options.reloadSdkSession).toHaveBeenCalledOnce();
await act(async () => {});
expect(drains).toHaveLength(2);
await act(async () => drains[1]?.({ status: "clean" }));
expect(options.reloadPreview).toHaveBeenCalledTimes(2);
expect(options.reloadSdkSession).toHaveBeenCalledTimes(2);
});

it("restores a durable unresolved conflict after remount", async () => {
Expand Down Expand Up @@ -285,4 +288,38 @@ describe("external file change coordinator", () => {
);
expect(onAcceptedPersistedFileChange).toHaveBeenCalledOnce();
});

it("completes a reload after a burst of rapid external writes", async () => {
const drains: Array<(result: { status: "clean" }) => void> = [];
const reloadPreview = vi.fn();
const onAcceptedPersistedFileChange = vi.fn();
await mountCoordinator({
drainPendingChanges: vi.fn(
() => new Promise<{ status: "clean" }>((resolve) => drains.push(resolve)),
),
reloadPreview,
onAcceptedPersistedFileChange,
});

// Fire three events in rapid succession (simulates generator + check + snapshot)
act(() => {
handler?.({ path: "index.html", content: "write-1", version: "v1" });
handler?.({ path: "index.html", content: "write-2", version: "v2" });
handler?.({ path: "index.html", content: "write-3", version: "v3" });
});

// Only one drain runs — events 2 and 3 are stashed (last one wins)
expect(drains).toHaveLength(1);

// Complete the first drain — triggers reload, then stashed event starts a second drain
await act(async () => drains[0]?.({ status: "clean" }));
expect(reloadPreview).toHaveBeenCalledOnce();
await act(async () => {});
expect(drains).toHaveLength(2);

// Complete the second drain — processes the final write
await act(async () => drains[1]?.({ status: "clean" }));
expect(reloadPreview).toHaveBeenCalledTimes(2);
expect(onAcceptedPersistedFileChange).toHaveBeenCalledTimes(2);
});
});
95 changes: 60 additions & 35 deletions packages/studio/src/hooks/useExternalFileChangeCoordinator.ts
Original file line number Diff line number Diff line change
Expand Up @@ -135,6 +135,8 @@ export function useExternalFileChangeCoordinator({
const lastEventIdentityRef = useRef<string | null>(null);
const blockedRef = useRef(blocked);
const snapshotWriteTailRef = useRef<Promise<void>>(Promise.resolve());
const drainingRef = useRef(false);
const pendingPayloadRef = useRef<{ payload: unknown } | null>(null);
blockedRef.current = blocked;

useEffect(() => {
Expand Down Expand Up @@ -213,37 +215,11 @@ export function useExternalFileChangeCoordinator({
await next;
}, []);

const processChange = useCallback(
const drainOnePending = useCallback(
// fallow-ignore-next-line complexity
async (payload: unknown, allowDuplicate = false) => {
async (payload: unknown) => {
const path = readStudioFileChangePath(payload);
if (!path || !projectId) return;
const pendingTimelinePaths = pendingTimelineEditPathRef.current;
// The old path-only suppression could drop a real agent/user write that
// raced ahead of the timeline write receipt. Clear the legacy marker but
// decide ownership only from the exact write token/content below.
pendingTimelinePaths.delete(path);

const content = readFileChangeContent(payload);
const token = readFileChangeWriteToken(payload);
logReload("file-change", { path, token: token ?? null, hasContent: content != null });
const identity = eventIdentity(path, payload);
if (!allowDuplicate && identity != null && identity === lastEventIdentityRef.current) {
logReload("suppressed", { path, why: "duplicate event" });
return;
}
lastEventIdentityRef.current = identity;

const ownWriteToken = consumeStudioWriteToken(token);
const ownContentEcho = content != null && isSelfWriteEcho(path, content);
if (ownWriteToken || ownContentEcho) {
onAcceptedPersistedFileChange(path);
logReload("suppressed", {
path,
why: ownWriteToken ? "own write token" : "own content echo",
});
return;
}
if (!path) return;

const generation = ++generationRef.current;
const result = await drainPendingChanges();
Expand All @@ -253,7 +229,7 @@ export function useExternalFileChangeCoordinator({
const previousBlocked = blockedRef.current;
if (previousBlocked?.status === "failed" && deleteConflictSnapshot) {
try {
await deleteConflictSnapshot(projectId, path);
await deleteConflictSnapshot(projectId!, path);
} catch (error) {
if (mountedRef.current && generation === generationRef.current) {
setBlocked({ ...previousBlocked, generation, error });
Expand All @@ -267,6 +243,7 @@ export function useExternalFileChangeCoordinator({
reloadAcceptedGeneration(path);
return;
}
const content = readFileChangeContent(payload);
if (result.status === "failed") {
const candidate = getPendingCandidate?.();
const studioContent = candidate?.path === path ? candidate.content : null;
Expand All @@ -275,7 +252,7 @@ export function useExternalFileChangeCoordinator({
try {
await persistSnapshotInOrder(() =>
persistFailureSnapshot(
projectId,
projectId!,
path,
studioContent,
readFileChangeVersion(payload),
Expand Down Expand Up @@ -305,7 +282,7 @@ export function useExternalFileChangeCoordinator({
return;
}
try {
await persistSnapshotInOrder(() => persistConflictSnapshot(projectId, result.error));
await persistSnapshotInOrder(() => persistConflictSnapshot(projectId!, result.error));
} catch (error) {
if (!mountedRef.current || generation !== generationRef.current) return;
setBlocked({
Expand All @@ -323,9 +300,8 @@ export function useExternalFileChangeCoordinator({
setBlocked({ status: "conflict", generation, error: result.error, payload });
},
[
projectId,
pendingTimelineEditPathRef,
drainPendingChanges,
projectId,
deleteConflictSnapshot,
getPendingCandidate,
persistConflictSnapshot,
Expand All @@ -336,6 +312,55 @@ export function useExternalFileChangeCoordinator({
],
);

const startDrainLoop = useCallback(async () => {
if (drainingRef.current) return;
drainingRef.current = true;
try {
while (mountedRef.current) {
const pending = pendingPayloadRef.current;
if (!pending) break;
pendingPayloadRef.current = null;
await drainOnePending(pending.payload);
}
} finally {
drainingRef.current = false;
}
}, [drainOnePending]);

const processChange = useCallback(
// fallow-ignore-next-line complexity
(payload: unknown) => {
const path = readStudioFileChangePath(payload);
if (!path || !projectId) return;
pendingTimelineEditPathRef.current.delete(path);

const content = readFileChangeContent(payload);
const token = readFileChangeWriteToken(payload);
logReload("file-change", { path, token: token ?? null, hasContent: content != null });
const identity = eventIdentity(path, payload);
if (identity != null && identity === lastEventIdentityRef.current) {
logReload("suppressed", { path, why: "duplicate event" });
return;
}
lastEventIdentityRef.current = identity;

const ownWriteToken = consumeStudioWriteToken(token);
const ownContentEcho = content != null && isSelfWriteEcho(path, content);
if (ownWriteToken || ownContentEcho) {
onAcceptedPersistedFileChange(path);
logReload("suppressed", {
path,
why: ownWriteToken ? "own write token" : "own content echo",
});
return;
}

pendingPayloadRef.current = { payload };
void startDrainLoop();
},
[projectId, pendingTimelineEditPathRef, startDrainLoop, onAcceptedPersistedFileChange],
);

useEffect(() => {
const handler = (payload?: unknown) => processChange(payload);
const adapter = testHotAdapter();
Expand All @@ -357,7 +382,7 @@ export function useExternalFileChangeCoordinator({
if (!current || current.status === "conflict" || current.recovered) return;
resetSaveQueues?.();
lastEventIdentityRef.current = null;
await processChange(current.payload, true);
processChange(current.payload);
}, [processChange, resetSaveQueues]);

const useExternalFile = useCallback(
Expand Down
Loading