From 7b0f592f254d076c111f5eb642109195b186905f Mon Sep 17 00:00:00 2001 From: Abdullah Sultan Date: Thu, 6 Aug 2026 12:01:54 -0500 Subject: [PATCH 1/6] Fall back to polling when the host cannot serve a stream `crosscode start` worked, published edits, and received none of its peers'. The daemon retried `GET /v1/stream` and got 404 on every attempt: the service runs as a Vercel serverless function, and a serverless function cannot answer a WebSocket upgrade. No amount of reconnecting was ever going to help, and there was no other path in -- so the product was an uploader, not a sync, and `crosscode status` reported `connected: false` with no way for it to ever become true. The receiving half already existed. `catchUp()` drains `GET /v1/changes?since=` and applies each change, and it already ran on every (re)subscribe to close the gap left by a dropped socket. It just had nothing to call it when the socket never opened at all. So a failed stream now starts a poll loop that calls the same catch-up every 3s, and stops the moment a socket opens. Two details worth naming: `connected` now means "in touch with the service by either transport", which is what `crosscode status` should report. Presence asks the new `streaming` instead, because presence exists only on the socket -- a polling room goes quiet about who is where, but it is not blind to edits. The poll loop guards re-entry on `pollActive` rather than on the timer handle. The timer is unset while a tick is in flight, and the socket fails once per backoff, so starting a second loop in that window would have been the normal case rather than the rare one. ## Tests `sync-service-stub` gained a `streams: false` option that omits the upgrade listener entirely, which is what makes node destroy the socket -- the client's view of a host that does not serve WebSockets. The new test stands two real daemons up against it and asserts an edit crosses in both directions. Verified it fails without the fix: it times out waiting for either daemon to report itself connected. `pnpm build` clean, `pnpm test` 215 passed / 11 skipped. Co-Authored-By: Claude Opus 5 --- apps/daemon/src/sync-daemon.test.ts | 24 ++++++++++ apps/daemon/src/sync-daemon.ts | 8 +++- apps/daemon/src/sync-service-client.ts | 64 +++++++++++++++++++++++++- apps/daemon/src/sync-service-stub.ts | 19 ++++++-- docs/protocol.md | 6 +++ 5 files changed, 114 insertions(+), 7 deletions(-) diff --git a/apps/daemon/src/sync-daemon.test.ts b/apps/daemon/src/sync-daemon.test.ts index 360272b..00650e1 100644 --- a/apps/daemon/src/sync-daemon.test.ts +++ b/apps/daemon/src/sync-daemon.test.ts @@ -219,6 +219,30 @@ describe("the daemon around the engine", () => { await waitFor("bob's cursor to advance", async () => bob.status().cursor > 0); }, 60_000); + /** + * The production outage this covers: the service runs as a serverless function, which + * cannot answer a WebSocket upgrade, so `/v1/stream` 404s forever. Publishing worked and + * receiving did not -- an uploader, not a sync -- and `crosscode status` said + * `connected: false` with no way for it to ever become true. + */ + it("still receives a peer's edits when the host cannot serve a stream at all", async () => { + const service = await startSyncServiceStub({ streams: false }); + services.push(service); + const origin = await seedOrigin({ "a.txt": "0\n" }); + const roots = await Promise.all(["alice", "bob"].map((name) => checkout(origin, name, service))); + const [alice, bob] = [await start(roots[0]!), await start(roots[1]!)]; + + // Reachable by the only transport there is, and honest about saying so. + await waitFor("both daemons to report themselves connected", async () => alice.status().connected && bob.status().connected); + + await type(roots[0]!, "a.txt", "arrived without a socket\n"); + await waitFor("bob to receive it", () => allEqual(roots, "a.txt", "arrived without a socket\n")); + + // And the other way, so it is a two-way sync rather than one lucky direction. + await type(roots[1]!, "a.txt", "and back again\n"); + await waitFor("alice to receive it", () => allEqual(roots, "a.txt", "and back again\n")); + }, 60_000); + it("resyncs from full content when the service says the cursor is too old", async () => { const service = await startSyncServiceStub(); services.push(service); diff --git a/apps/daemon/src/sync-daemon.ts b/apps/daemon/src/sync-daemon.ts index 29d0d95..2805a22 100644 --- a/apps/daemon/src/sync-daemon.ts +++ b/apps/daemon/src/sync-daemon.ts @@ -231,7 +231,11 @@ export class SyncDaemon { // Announce on every (re)subscribe: a peer that joins after we committed would // otherwise never learn our HEAD, since presence is only sent when something happens. onConnected: () => { void this.enqueue(async () => { await this.catchUp(); await this.announce(); }); }, - onDisconnected: () => {} + onDisconnected: () => {}, + // The stream's fallback. Same catch-up, on a timer instead of on a subscribe, so a + // checkout that can never open a socket still receives its peers' edits. Presence is + // not sent here: it lives only on the socket, so the room goes quiet but not blind. + onPoll: () => this.enqueue(() => this.catchUp()) }, () => this.cursor); } @@ -371,7 +375,7 @@ export class SyncDaemon { /** Tells the room where this checkout is. Silent when there is no stream to say it on. */ private async announce(paths: string[] = []): Promise { - if (!this.client.id || !this.client.connected) return; + if (!this.client.id || !this.client.streaming) return; await this.client.presence(paths, this.options.actor ?? "you", this.head ?? undefined).catch(() => {}); } diff --git a/apps/daemon/src/sync-service-client.ts b/apps/daemon/src/sync-service-client.ts index a7ec79b..2fb3ac3 100644 --- a/apps/daemon/src/sync-service-client.ts +++ b/apps/daemon/src/sync-service-client.ts @@ -15,6 +15,14 @@ import { /** * The daemon's half of the sync wire contract: four routes and one socket. Nothing here * knows what a shadow ref is -- it moves `FileVersion`s and hands back `Change`s. + * + * The socket is the fast path, not the only path. Where the service runs as a serverless + * function it cannot answer a WebSocket upgrade at all, so `/v1/stream` 404s on every + * attempt and no amount of reconnecting will ever help. Without a fallback that leaves the + * daemon publishing happily and receiving nothing -- an uploader, not a sync. So a failed + * stream falls back to polling `/v1/changes?since=`, the same route the daemon already + * uses to catch up after a restart. Slower, and presence goes quiet because it only exists + * on the socket, but edits still arrive. */ export type SyncStreamHandlers = { @@ -23,14 +31,23 @@ export type SyncStreamHandlers = { /** Fired on every (re)subscribe: the daemon catches up from its cursor here. */ onConnected?: () => void; onDisconnected?: () => void; + /** + * Fired on each poll while the stream is unavailable. Resolve to report the service was + * reachable, reject to report it was not; the client reports that as `connected`. + */ + onPoll?: () => Promise; }; const INITIAL_BACKOFF_MS = 500; const MAX_BACKOFF_MS = 15_000; +const POLL_INTERVAL_MS = 3_000; export class SyncServiceClient { private socket?: WebSocket; private reconnectTimer?: NodeJS.Timeout; + private pollTimer?: NodeJS.Timeout; + private pollActive = false; + private polling = false; private backoffMs = INITIAL_BACKOFF_MS; private stopped = false; private handlers?: SyncStreamHandlers; @@ -42,7 +59,13 @@ export class SyncServiceClient { private replicaId?: string ) {} + /** In touch with the service, by either transport -- what `crosscode status` reports. */ get connected(): boolean { + return this.streaming || this.polling; + } + + /** The socket specifically. Presence exists only here, so it asks this and not `connected`. */ + get streaming(): boolean { return this.socket?.readyState === WebSocket.OPEN; } @@ -78,7 +101,7 @@ export class SyncServiceClient { } async presence(paths: string[], actor: string, head?: string): Promise { - if (!this.connected) return; + if (!this.streaming) return; const message = { type: "presence" as const, peers: [presenceSchema.parse({ replicaId: this.requireReplicaId(), actor, branch: this.branch, paths: paths.slice(0, 50), ...(head ? { head } : {}) })] }; this.socket!.send(JSON.stringify(message)); } @@ -94,6 +117,8 @@ export class SyncServiceClient { stop(): void { this.stopped = true; if (this.reconnectTimer) clearTimeout(this.reconnectTimer); + if (this.pollTimer) clearTimeout(this.pollTimer); + this.polling = false; const socket = this.socket; this.socket = undefined; socket?.close(); @@ -112,6 +137,7 @@ export class SyncServiceClient { since: this.cursorProvider() })); this.backoffMs = INITIAL_BACKOFF_MS; + this.stopPolling(); this.handlers?.onConnected?.(); }); socket.on("message", (data) => { @@ -127,6 +153,9 @@ export class SyncServiceClient { if (this.socket === socket) this.socket = undefined; this.handlers?.onDisconnected?.(); this.scheduleReconnect(); + // The socket is down for an unknown length of time -- one dropped frame, or a host + // that will never serve one. Polling covers both, and stops as soon as one opens. + this.startPolling(); }); socket.on("error", () => socket.close()); } @@ -139,6 +168,39 @@ export class SyncServiceClient { this.reconnectTimer.unref(); } + /** + * Self-scheduling rather than an interval, so a slow catch-up never overlaps itself. + * `pollActive` is separate from `pollTimer` because the timer is unset while a tick is in + * flight, and the socket fails often enough -- once per backoff -- that starting a second + * loop in that window would be the normal case, not the rare one. + */ + private startPolling(): void { + if (this.pollActive || this.stopped || !this.handlers?.onPoll) return; + this.pollActive = true; + const tick = async () => { + this.pollTimer = undefined; + if (this.stopped || this.streaming) { this.stopPolling(); return; } + try { + await this.handlers?.onPoll?.(); + this.polling = true; + } catch { + // Unreachable by either transport, which is what `connected: false` should mean. + this.polling = false; + } + if (this.stopped || this.streaming) { this.stopPolling(); return; } + this.pollTimer = setTimeout(() => void tick(), POLL_INTERVAL_MS); + this.pollTimer.unref(); + }; + void tick(); + } + + private stopPolling(): void { + if (this.pollTimer) clearTimeout(this.pollTimer); + this.pollTimer = undefined; + this.pollActive = false; + this.polling = false; + } + private requireReplicaId(): string { if (!this.replicaId) throw new Error("This replica is not registered yet"); return this.replicaId; diff --git a/apps/daemon/src/sync-service-stub.ts b/apps/daemon/src/sync-service-stub.ts index 9a5f5a7..d7b59f9 100644 --- a/apps/daemon/src/sync-service-stub.ts +++ b/apps/daemon/src/sync-service-stub.ts @@ -35,7 +35,16 @@ export type SyncServiceStub = { close(): Promise; }; -export async function startSyncServiceStub(): Promise { +export type SyncServiceStubOptions = { + /** + * Whether the stub answers a WebSocket upgrade at all. False stands in for a serverless + * host, which cannot: the routes work, `/v1/stream` never will, and the daemon has to + * fall back to polling rather than reconnect forever. + */ + streams?: boolean; +}; + +export async function startSyncServiceStub(options: SyncServiceStubOptions = {}): Promise { const changes: Change[] = []; const subscribers = new Set(); let sequence = 0; @@ -113,8 +122,10 @@ export async function startSyncServiceStub(): Promise { } }; - const sockets = new WebSocketServer({ server, path: "/v1/stream" }); - sockets.on("connection", (socket) => { + // No upgrade listener at all when streams are off, which is what makes node destroy the + // socket -- the client's view of a host that does not serve WebSockets. + const sockets = options.streams === false ? undefined : new WebSocketServer({ server, path: "/v1/stream" }); + sockets?.on("connection", (socket) => { let subscriber: Subscriber | undefined; socket.on("message", (data) => { const message: unknown = JSON.parse(data.toString()); @@ -153,7 +164,7 @@ export async function startSyncServiceStub(): Promise { }, async close() { for (const subscriber of [...subscribers]) subscriber.socket.terminate(); - sockets.close(); + sockets?.close(); // `server.close` only stops new connections; it waits for the open ones, and the // daemon's fetch keeps its sockets alive between requests. Without this the close // never resolves and the test's teardown hook dies on its own timeout instead. diff --git a/docs/protocol.md b/docs/protocol.md index 6ce6ec5..63b0aca 100644 --- a/docs/protocol.md +++ b/docs/protocol.md @@ -92,6 +92,12 @@ The client sends `subscribe` (`projectId`, `branch`, `replicaId`, `since`) and t `replicaId`, `actor`, `branch`, and up to 50 recently touched `paths`, which is how an agent can answer "who is working on what" without any ambient UI. +The stream is the fast path, not the only one. A host that cannot serve a WebSocket upgrade +— a serverless function, for one — leaves a daemon publishing happily and receiving +nothing, so a daemon that cannot open the stream falls back to polling `GET /v1/changes?since=` +every few seconds. Edits still arrive, a few seconds later than they would have. Presence +does not: it exists only on the socket, so a polling room goes quiet about who is where. + ## Daemon config and local API `syncDaemonConfig` is what the daemon stores per checkout: `projectId`, `repo`, and the From 31c8a936b37abbaaac454432504c9a5bc5bc5423 Mon Sep 17 00:00:00 2001 From: Abdullah Sultan Date: Thu, 6 Aug 2026 12:40:22 -0500 Subject: [PATCH 2/6] Keep the session alive, and stop reporting a daemon that is about to die Three defects found by running the workflow against production rather than against the stub. Each was hidden behind the last, and together they made a checkout unusable roughly an hour after sign-in with no way to recover from inside the CLI. ## The session was never refreshed Supabase access tokens last an hour. The refresh token sat in the config next to the expired access token and nothing ever spent it, so `register()` threw "Access token is invalid or expired" and took the whole daemon down. Every authenticated call now goes through `authorization()`, which refreshes a minute ahead of expiry and hands the refreshed session back to be written to disk -- Supabase rotates the refresh token on use, so persisting it is what makes the *next* start-up work. Concurrent callers share one refresh through a promise, because two would each spend a token the other had already rotated away. ## `start` reported a daemon that was already dead `crosscode start` polls for the descriptor every 100ms and reports success the moment it appears. It was written when the loopback server bound -- before the stream, the watcher, or the first publish -- so a daemon that then died on the expired session still had `start` printing "Daemon started; this checkout is syncing" about a process that was gone a second later. The descriptor is now published last, once start-up has actually succeeded. ## And there was no way out of the loop The daemon told the user to run `crosscode start`. `start` saw a session on disk, said "Already signed in for this checkout", and started the same doomed daemon again. "There is a session" and "this checkout can talk to the service" are different questions; `start` now asks the second one, refreshing if it can and signing in again if it cannot. An expired sign-in also exits with that one line instead of a stack trace. ## Tests - The refresh path: refreshes when expired, leaves a live token alone, and spends the refresh token exactly once under three concurrent calls. - The descriptor: against a service that accepts and never answers, no descriptor exists while start-up is in flight. Verified it fails without the fix. - `start`: signs in again when the refresh fails, and reuses (and stores) a refreshed session without a second sign-in. `streamPollMs` joins the other tunable intervals so the polling test does not wait out real 3s cycles. `pnpm build` clean, `pnpm test` 219 passed / 11 skipped. Two tests in `sync-daemon.test.ts` flake under full-suite CPU contention on a loaded machine -- always as timeouts, never assertions. Confirmed pre-existing: `main` fails the same way at the same rate with none of this applied. Co-Authored-By: Claude Opus 5 --- apps/cli/src/index.test.ts | 1 + apps/cli/src/setup.test.ts | 51 ++++++++++-- apps/cli/src/setup.ts | 35 +++++++-- apps/daemon/src/main.ts | 8 ++ apps/daemon/src/supabase-client.ts | 21 +++++ apps/daemon/src/sync-daemon.test.ts | 37 +++++++++ apps/daemon/src/sync-daemon.ts | 36 ++++++++- apps/daemon/src/sync-service-client.test.ts | 87 +++++++++++++++++++++ apps/daemon/src/sync-service-client.ts | 56 ++++++++++--- 9 files changed, 307 insertions(+), 25 deletions(-) create mode 100644 apps/daemon/src/sync-service-client.test.ts diff --git a/apps/cli/src/index.test.ts b/apps/cli/src/index.test.ts index 02360d9..8942b81 100644 --- a/apps/cli/src/index.test.ts +++ b/apps/cli/src/index.test.ts @@ -35,6 +35,7 @@ function environmentWithDaemon(daemon: Partial): Environment { serviceUrl: "https://www.getcrosscode.dev", createService: unused, signIn: unused, + refreshSession: unused, createDaemon: () => ({ start: async () => ({ alreadyRunning: false, pid: 1 }), stop: async () => ({ wasRunning: false }), diff --git a/apps/cli/src/setup.test.ts b/apps/cli/src/setup.test.ts index 0ae60ec..b15348f 100644 --- a/apps/cli/src/setup.test.ts +++ b/apps/cli/src/setup.test.ts @@ -9,7 +9,7 @@ import { syncDaemonConfigSchema } from "../../../packages/protocol/src/sync.js"; // two sides agree, so asserting it with the CLI's helper alone would prove nothing. import { readSyncConfig, syncConfigPath } from "../../daemon/src/sync-config.js"; import { installAgentSurface } from "./agent-surface.js"; -import { configPath, readConfig } from "./config.js"; +import { configPath, readConfig, writeConfig } from "./config.js"; import type { DaemonControl } from "./daemon.js"; import { httpSyncService, type SyncService } from "./service.js"; import { setup, type Environment } from "./setup.js"; @@ -37,8 +37,8 @@ const SESSION = { accessToken: "token", refreshToken: "refresh", expiresAt: "203 * the daemon -- each counting its calls, because "idempotent" here means exactly "called * once no matter how many times start runs". */ -function stubEnvironment(overrides: { service?: Partial } = {}) { - const calls = { createProject: 0, redeemInvite: 0, signIn: 0, daemonStart: 0 }; +function stubEnvironment(overrides: { service?: Partial; refreshSession?: Environment["refreshSession"] } = {}) { + const calls = { createProject: 0, redeemInvite: 0, signIn: 0, daemonStart: 0, refreshSession: 0 }; let daemonRunning = false; const service: SyncService = { @@ -74,6 +74,10 @@ function stubEnvironment(overrides: { service?: Partial } = {}) { calls.signIn += 1; return { session: SESSION, githubToken: "gho_test" }; }, + refreshSession: async (session) => { + calls.refreshSession += 1; + return overrides.refreshSession ? overrides.refreshSession(session) : session; + }, createDaemon: () => daemon, // The real installer, not a stub: whether installing twice duplicates anything is the // question, so it has to be the code that writes the files. @@ -90,7 +94,7 @@ describe("crosscode start", () => { const result = await setup(root, environment); expect(result).toMatchObject({ repoRoot: root, repo: "acme/app", projectId: "project-1", signedIn: "just-now", project: "created" }); - expect(calls).toEqual({ createProject: 1, redeemInvite: 0, signIn: 1, daemonStart: 1 }); + expect(calls).toEqual({ createProject: 1, redeemInvite: 0, signIn: 1, daemonStart: 1, refreshSession: 0 }); // What it wrote is the contract's own shape, not a CLI-private one. expect(syncDaemonConfigSchema.parse(JSON.parse(await readFile(await configPath(root), "utf8")))).toMatchObject({ projectId: "project-1", @@ -117,6 +121,41 @@ describe("crosscode start", () => { expect(await readSyncConfig(root)).toMatchObject({ projectId: "project-1", repo: "acme/app" }); }); + /** + * The dead end this closes. Supabase access tokens last an hour and refresh tokens are + * spent on first use, so a checkout comes back to a session that is on disk and useless. + * The daemon died on it saying "run `crosscode start`"; `start` saw a session, said + * "Already signed in for this checkout", and started the same doomed daemon again. There + * was no way out of that loop from inside the CLI. + */ + it("signs in again when the stored session cannot be refreshed", async () => { + const root = await checkout(); + const { environment, calls } = stubEnvironment({ refreshSession: async () => undefined }); + await setup(root, environment); + await writeConfig(root, { ...(await readSyncConfig(root)), service: { url: environment.serviceUrl, session: { ...SESSION, expiresAt: "2020-01-01T00:00:00.000Z" } } }); + + const second = await setup(root, environment); + + expect(calls.refreshSession).toBe(1); + expect(second.signedIn).toBe("just-now"); + expect(calls.signIn).toBe(2); + }); + + it("reuses a session it can still refresh, without a second sign-in", async () => { + const root = await checkout(); + const refreshed = { ...SESSION, accessToken: "refreshed", expiresAt: "2031-01-01T00:00:00.000Z" }; + const { environment, calls } = stubEnvironment({ refreshSession: async () => refreshed }); + await setup(root, environment); + await writeConfig(root, { ...(await readSyncConfig(root)), service: { url: environment.serviceUrl, session: { ...SESSION, expiresAt: "2020-01-01T00:00:00.000Z" } } }); + + const second = await setup(root, environment); + + expect(second.signedIn).toBe("already"); + expect(calls.signIn).toBe(1); + // The refreshed token is what the daemon will be started with, so it has to be stored. + expect((await readSyncConfig(root)).service.session?.accessToken).toBe("refreshed"); + }); + // The property the invite flow depends on: the person sending the link has already run // start, and the person receiving it is told to run the same command. it("is idempotent -- one project, one daemon, no duplicated MCP, skill, or hook install", async () => { @@ -126,7 +165,7 @@ describe("crosscode start", () => { const first = await setup(root, environment); const second = await setup(root, environment); - expect(calls).toEqual({ createProject: 1, redeemInvite: 0, signIn: 1, daemonStart: 1 }); + expect(calls).toEqual({ createProject: 1, redeemInvite: 0, signIn: 1, daemonStart: 1, refreshSession: 0 }); expect(second.projectId).toBe(first.projectId); expect(second).toMatchObject({ signedIn: "already", project: "existing", daemon: { alreadyRunning: true } }); expect(second.agent).toMatchObject({ mcp: { changed: false }, skill: { changed: false }, hooks: { changed: false } }); @@ -191,7 +230,7 @@ describe("crosscode join ", () => { const result = await setup(root, environment, { code: "CC-7F3A-9C2E" }); - expect(calls).toEqual({ createProject: 0, redeemInvite: 1, signIn: 1, daemonStart: 1 }); + expect(calls).toEqual({ createProject: 0, redeemInvite: 1, signIn: 1, daemonStart: 1, refreshSession: 0 }); expect(result).toMatchObject({ projectId: "project-from-invite", repo: "acme/app", project: "joined", daemon: { alreadyRunning: false } }); expect(await readConfig(root)).toMatchObject({ projectId: "project-from-invite", repo: "acme/app", service: { session: SESSION } }); expect(result.agent).toMatchObject({ mcp: { changed: true }, skill: { changed: true }, hooks: { changed: true } }); diff --git a/apps/cli/src/setup.ts b/apps/cli/src/setup.ts index 97b6077..7ddb58a 100644 --- a/apps/cli/src/setup.ts +++ b/apps/cli/src/setup.ts @@ -7,6 +7,7 @@ import { readConfig, writeConfig } from "./config.js"; import { localDaemon, type DaemonControl } from "./daemon.js"; import { CliError } from "./errors.js"; import { httpSyncService, type SyncService } from "./service.js"; +import { refreshStoredSession, type StoredSession } from "../../daemon/src/supabase-client.js"; import { SERVICE_URL } from "./version.js"; /** @@ -26,6 +27,8 @@ export type Environment = { serviceUrl: string; createService(accessToken: string): SyncService; signIn: SignIn; + /** Trades a refresh token for a live one, or resolves undefined if it cannot. */ + refreshSession(session: StoredSession): Promise; createDaemon(repoRoot: string): DaemonControl; installAgentSurface(repoRoot: string): Promise; }; @@ -35,6 +38,7 @@ export function defaultEnvironment(): Environment { serviceUrl: SERVICE_URL, createService: (accessToken) => httpSyncService(SERVICE_URL, accessToken), signIn: githubSignIn(), + refreshSession: (session) => refreshStoredSession(session).catch(() => undefined), createDaemon: (repoRoot) => localDaemon(repoRoot), installAgentSurface: (repoRoot) => installAgentSurface(repoRoot) }; @@ -79,12 +83,20 @@ export async function setup(directory: string, environment: Environment, options // so it exists for exactly as long as the handshake that produced it. `join` therefore // signs in even in a checkout that is already signed in, rather than failing at the // redeem call with nothing to offer. - const reuseSession = session && !options.code; - if (reuseSession) report("Already signed in for this checkout."); - const signedIn = reuseSession + // + // A stored session is only worth reusing if it still works. Supabase access tokens last + // an hour and refresh tokens are spent on first use, so "there is a session on disk" and + // "this checkout can talk to the service" are different questions. Answering the first + // one left the user with no way out: the daemon died on the dead session telling them to + // run `crosscode start`, and `start` said "Already signed in for this checkout" and + // started the same doomed daemon again. + const usable = session && !options.code ? await usableSession(session, environment) : undefined; + if (usable) report("Already signed in for this checkout."); + else if (session && !options.code) report("This checkout's sign-in has expired; signing in again."); + const signedIn = usable ? undefined : await environment.signIn({ serviceUrl: environment.serviceUrl, openBrowser: options.openBrowser !== false, report }); - const activeSession = signedIn?.session ?? session!; + const activeSession = signedIn?.session ?? usable!; const service = environment.createService(activeSession.accessToken); const { projectId, repo, origin } = await resolveProject(service, existing, localRepo, repository.root, options.code, signedIn?.githubToken, report); @@ -105,13 +117,26 @@ export async function setup(directory: string, environment: Environment, options repo, projectId, branch: repository.branch, - signedIn: reuseSession ? "already" : "just-now", + signedIn: usable ? "already" : "just-now", project: origin, daemon, agent }; } +/** + * The stored session if it can still be used, refreshed first if it is close to expiring, + * and undefined if the refresh token is spent or revoked -- in which case the only way + * forward is signing in again, which is what the caller does. + */ +async function usableSession(session: StoredSession, environment: Environment): Promise { + if (Date.parse(session.expiresAt) - Date.now() > EXPIRY_MARGIN_MS) return session; + return environment.refreshSession(session); +} + +/** Refresh this far ahead of expiry, so setup never hands the daemon a token about to die. */ +const EXPIRY_MARGIN_MS = 60_000; + /** * Which project this checkout syncs, in the one order that keeps `start` idempotent: * an invite code wins, then whatever this checkout already chose, then a new project. diff --git a/apps/daemon/src/main.ts b/apps/daemon/src/main.ts index 0a28f5e..8f4ab1f 100644 --- a/apps/daemon/src/main.ts +++ b/apps/daemon/src/main.ts @@ -1,6 +1,7 @@ #!/usr/bin/env node import { spawn } from "node:child_process"; import { createDaemonTelemetry } from "./observability.js"; +import { SessionExpiredError } from "./supabase-client.js"; import { supervise } from "./supervisor.js"; import { SyncDaemon } from "./sync-daemon.js"; import { VERSION } from "./version.js"; @@ -33,6 +34,13 @@ if (!directory) { const daemon = await SyncDaemon.start(directory).catch(async (error: unknown) => { telemetry.capture(error, "startup"); await telemetry.flush(); + // An expired sign-in is the one start-up failure with a one-line fix, and a stack trace + // buries it. Everything else still throws, because it is a bug and the trace is the + // useful part. + if (error instanceof SessionExpiredError) { + process.stderr.write(`${error.message}\n`); + process.exit(1); + } throw error; }); process.stdout.write(`${JSON.stringify({ ready: true, pid: process.pid, port: daemon.connection.port })}\n`); diff --git a/apps/daemon/src/supabase-client.ts b/apps/daemon/src/supabase-client.ts index 3e78125..9d81071 100644 --- a/apps/daemon/src/supabase-client.ts +++ b/apps/daemon/src/supabase-client.ts @@ -70,6 +70,27 @@ export function getSupabaseClient(): SupabaseClient { return createClient(url, anonKey, { auth: { persistSession: false, autoRefreshToken: false } }); } +/** Re-signing-in is the only fix, so this is worth telling apart from a network failure. */ +export class SessionExpiredError extends Error {} + +/** + * Trades a refresh token for a fresh access token. Supabase access tokens last an hour, so + * without this a daemon that has been running since before lunch holds a token the service + * refuses -- which is exactly what happened: `register()` threw "Access token is invalid or + * expired" and took the process down, and the only way back was signing in again. + */ +export async function refreshStoredSession(session: StoredSession): Promise { + const supabase = getSupabaseClient(); + const { data, error } = await supabase.auth.refreshSession({ refresh_token: session.refreshToken }); + if (error || !data.session) { + // A refresh token is spent on first use and rotated, so "already used" and "expired" + // both mean the same thing to the person reading this: sign in again. Say that, rather + // than leaving them with Supabase's wording and a stack trace. + throw new SessionExpiredError(`This checkout's Crosscode sign-in is no longer valid (${error?.message ?? "no session returned"}); run \`crosscode start\` to sign in again`); + } + return toStoredSession(data.session); +} + export function toStoredSession(session: Session): StoredSession { const expiresAtMs = session.expires_at ? session.expires_at * 1_000 : Date.now() + (session.expires_in ?? 3_600) * 1_000; return { accessToken: session.access_token, refreshToken: session.refresh_token, expiresAt: new Date(expiresAtMs).toISOString() }; diff --git a/apps/daemon/src/sync-daemon.test.ts b/apps/daemon/src/sync-daemon.test.ts index 00650e1..5fb96e9 100644 --- a/apps/daemon/src/sync-daemon.test.ts +++ b/apps/daemon/src/sync-daemon.test.ts @@ -1,4 +1,5 @@ import { execFile } from "node:child_process"; +import { createServer } from "node:http"; import { mkdtemp, mkdir, readFile, rm, writeFile } from "node:fs/promises"; import { tmpdir } from "node:os"; import { dirname, join } from "node:path"; @@ -36,6 +37,7 @@ const FAST: SyncDaemonOptions = { flushMs: 100, deferralMs: 60, gitPollMs: 100, + streamPollMs: 100, // The 10s hot window is real behaviour, not a constant worth waiting out in a test. engine: { hotWindowMs: 150, maxDeferrals: 3 } }; @@ -243,6 +245,41 @@ describe("the daemon around the engine", () => { await waitFor("alice to receive it", () => allEqual(roots, "a.txt", "and back again\n")); }, 60_000); + /** + * `crosscode start` polls for the descriptor every 100ms and reports success the moment + * it appears. Written when the loopback server bound, it appeared before the stream was + * up -- so a daemon that then died on an expired session still had `start` announcing + * "Daemon started; this checkout is syncing" about a process that was already gone. + */ + it("does not advertise itself as ready while start-up is still in flight", async () => { + // Accepts the connection and never answers, so the daemon is unambiguously still + // starting up while the assertions below run. + const hanging = createServer(() => {}); + await new Promise((ready) => hanging.listen(0, "127.0.0.1", ready)); + const service = await startSyncServiceStub(); + services.push(service); + const origin = await seedOrigin({ "a.txt": "0\n" }); + const root = await checkout(origin, "alice", service); + await writeSyncConfig(root, { + projectId: "project-1", repo: "acme/app", + service: { url: `http://127.0.0.1:${(hanging.address() as { port: number }).port}` } + }); + + const starting = start(root).catch(() => undefined); + // Comfortably past the point the loopback server has bound and the old code would have + // published the descriptor, and well inside the request that never returns. + await new Promise((settle) => setTimeout(settle, 2_000)); + const descriptor = await read(root, ".git/crosscode/daemon.json"); + + // Dropping the connection fails the pending request now rather than at its 10s timeout, + // which is 8s this suite does not need to spend. + hanging.closeAllConnections(); + await new Promise((closed) => hanging.close(() => closed())); + await starting; + + expect(descriptor).toBeNull(); + }, 60_000); + it("resyncs from full content when the service says the cursor is too old", async () => { const service = await startSyncServiceStub(); services.push(service); diff --git a/apps/daemon/src/sync-daemon.ts b/apps/daemon/src/sync-daemon.ts index 2805a22..18c95ac 100644 --- a/apps/daemon/src/sync-daemon.ts +++ b/apps/daemon/src/sync-daemon.ts @@ -21,9 +21,11 @@ import { readSyncState, removeOwnedDescriptor, writeDaemonDescriptor, + writeSyncConfig, writeSyncState, type DaemonDescriptor } from "./sync-config.js"; +import type { StoredSession } from "./supabase-client.js"; import { SyncServiceClient } from "./sync-service-client.js"; /** @@ -44,6 +46,8 @@ export type SyncDaemonOptions = { deferralMs?: number; /** How often the daemon looks for a branch switch or an in-progress git operation. */ gitPollMs?: number; + /** How often changes are pulled when the stream is unavailable. Real default is 3s. */ + streamPollMs?: number; port?: number; actor?: string; engine?: EngineOptions; @@ -67,6 +71,8 @@ export class SyncDaemon { private watcher?: FSWatcher; private server?: Server; private descriptor?: DaemonDescriptor; + /** Built when the loopback server binds, published only once start-up has fully succeeded. */ + private pendingDescriptor?: DaemonDescriptor; private timers: NodeJS.Timeout[] = []; private readonly debounce = new Map(); private readonly dirty = new Set(); @@ -103,7 +109,7 @@ export class SyncDaemon { const state = await readSyncState(root); const branch = await currentBranch(root); const engine = await SyncEngine.open(root, options.engine ?? {}); - const client = new SyncServiceClient(config, branch, state.branch === branch ? state.replicaId : undefined); + const client = new SyncServiceClient(config, branch, state.branch === branch ? state.replicaId : undefined, persistSession(root, config), options.streamPollMs); const daemon = new SyncDaemon(root, config, engine, client, branch, await headCommit(root), { debounceMs: options.debounceMs ?? 300, flushMs: options.flushMs ?? 2_000, @@ -127,6 +133,7 @@ export class SyncDaemon { this.timers.push(interval(() => void this.enqueue(() => this.observeGit()), this.options.gitPollMs)); // Anything edited while the daemon was down is still a change the peers need. await this.enqueue(() => this.publishAll()); + await this.announceReady(); } // ---- serialization ---------------------------------------------------- @@ -335,7 +342,7 @@ export class SyncDaemon { this.headNotice = undefined; this.reviewedHeads = ""; this.client.stop(); - this.client = new SyncServiceClient(this.config, branch); + this.client = new SyncServiceClient(this.config, branch, undefined, persistSession(this.root, this.config), this.options.streamPollMs); if (this.options.stream ?? true) await this.startStream(); await this.publishAll(); return; @@ -470,12 +477,24 @@ export class SyncDaemon { server.listen(this.options.port ?? 0, "127.0.0.1", () => { server.off("error", failed); ready(); }); }); this.server = server; - this.descriptor = daemonDescriptorSchema.parse({ + this.pendingDescriptor = daemonDescriptorSchema.parse({ pid: process.pid, port: (server.address() as { port: number }).port, secret, startedAt: new Date().toISOString() }); + } + + /** + * The descriptor is the daemon saying "I am up", so it is written last -- after the + * stream, the watcher, and the first publish have all succeeded. Written from startHttp() + * as it used to be, a daemon that then died on an expired session still left `crosscode + * start` reporting "Daemon started; this checkout is syncing" about a process that was + * gone a second later, because `start` polls for exactly this file. + */ + private async announceReady(): Promise { + if (!this.pendingDescriptor) throw new Error("The daemon's loopback server was never started"); + this.descriptor = this.pendingDescriptor; await writeDaemonDescriptor(this.root, this.descriptor); } @@ -561,6 +580,17 @@ function interval(task: () => void, ms: number): NodeJS.Timeout { const MAX_BODY_BYTES = 4 * 1024 * 1024; +/** + * Writes a refreshed session back to the checkout's config. Supabase rotates the refresh + * token on every use, so this is not an optimization: without it a restart would present + * the token the last refresh already spent, and sign-in would be the only way back. + */ +function persistSession(root: string, config: SyncDaemonConfig): (session: StoredSession) => Promise { + return async (session) => { + await writeSyncConfig(root, { ...config, service: { ...config.service, session } }); + }; +} + async function readJsonBody(request: import("node:http").IncomingMessage): Promise { if (request.headers["content-type"]?.split(";")[0]?.trim() !== "application/json") throw new HttpError(415, "Unsupported content type"); const chunks: Buffer[] = []; diff --git a/apps/daemon/src/sync-service-client.test.ts b/apps/daemon/src/sync-service-client.test.ts new file mode 100644 index 0000000..7aec77c --- /dev/null +++ b/apps/daemon/src/sync-service-client.test.ts @@ -0,0 +1,87 @@ +import { createServer, type Server } from "node:http"; +import type { AddressInfo } from "node:net"; +import { afterEach, describe, expect, it, vi } from "vitest"; +import type { StoredSession } from "./supabase-client.js"; + +/** + * Supabase access tokens last an hour and a daemon is expected to outlive that. Before + * this, one did not: `register()` threw "Access token is invalid or expired", the daemon + * died with a stack trace, and the refresh token sitting in the config next to the dead + * access token was never spent. + */ + +const refreshStoredSession = vi.hoisted(() => vi.fn()); +vi.mock("./supabase-client.js", () => ({ refreshStoredSession })); + +const { SyncServiceClient } = await import("./sync-service-client.js"); + +const servers: Server[] = []; +afterEach(async () => { + refreshStoredSession.mockReset(); + await Promise.all(servers.splice(0).map((server) => new Promise((resolve) => server.close(() => resolve())))); +}); + +/** Records the bearer on every request, and answers whatever the client asked for. */ +async function recordingService(): Promise<{ url: string; bearers: string[] }> { + const bearers: string[] = []; + const server = createServer((request, response) => { + bearers.push(String(request.headers.authorization)); + response.writeHead(200, { "content-type": "application/json" }); + response.end(JSON.stringify({ ok: true, data: { replicaId: "11111111-2222-4333-8444-555566667777", cursor: 0 } })); + }); + servers.push(server); + await new Promise((ready) => server.listen(0, "127.0.0.1", ready)); + return { url: `http://127.0.0.1:${(server.address() as AddressInfo).port}`, bearers }; +} + +const session = (expiresAt: string, accessToken = "old-token"): StoredSession => + ({ accessToken, refreshToken: "refresh-1", expiresAt }); + +const configFor = (url: string, expiresAt: string) => + ({ projectId: "project-1", repo: "acme/app", service: { url, session: session(expiresAt) } }); + +const inAnHour = () => new Date(Date.now() + 3_600_000).toISOString(); +const anHourAgo = () => new Date(Date.now() - 3_600_000).toISOString(); + +describe("the sync client's session", () => { + it("refreshes an expired token, persists it, and sends the new one", async () => { + const service = await recordingService(); + refreshStoredSession.mockResolvedValue(session(inAnHour(), "new-token")); + const persisted: StoredSession[] = []; + const client = new SyncServiceClient(configFor(service.url, anHourAgo()), "main", undefined, async (next) => { persisted.push(next); }); + + await client.register(); + + expect(refreshStoredSession).toHaveBeenCalledTimes(1); + expect(service.bearers).toEqual(["Bearer new-token"]); + // Persisted, because Supabase rotates the refresh token: the next start-up needs this + // one, not the one that was already spent. + expect(persisted.map((entry) => entry.accessToken)).toEqual(["new-token"]); + }); + + it("leaves a token that is still good alone", async () => { + const service = await recordingService(); + const client = new SyncServiceClient(configFor(service.url, inAnHour()), "main"); + + await client.register(); + + expect(refreshStoredSession).not.toHaveBeenCalled(); + expect(service.bearers).toEqual(["Bearer old-token"]); + }); + + it("spends the refresh token once when several calls find it expired together", async () => { + const service = await recordingService(); + // Supabase rotates the refresh token, so a second concurrent refresh would be spending + // one that no longer exists. + refreshStoredSession.mockImplementation(async () => { + await new Promise((resolve) => setTimeout(resolve, 20)); + return session(inAnHour(), "new-token"); + }); + const client = new SyncServiceClient(configFor(service.url, anHourAgo()), "main"); + + await Promise.all([client.register(), client.register(), client.register()]); + + expect(refreshStoredSession).toHaveBeenCalledTimes(1); + expect(service.bearers).toEqual(["Bearer new-token", "Bearer new-token", "Bearer new-token"]); + }); +}); diff --git a/apps/daemon/src/sync-service-client.ts b/apps/daemon/src/sync-service-client.ts index 2fb3ac3..d3569bb 100644 --- a/apps/daemon/src/sync-service-client.ts +++ b/apps/daemon/src/sync-service-client.ts @@ -1,4 +1,5 @@ import { WebSocket } from "ws"; +import { refreshStoredSession, type StoredSession } from "./supabase-client.js"; import { changesResponseSchema, presenceSchema, @@ -41,6 +42,8 @@ export type SyncStreamHandlers = { const INITIAL_BACKOFF_MS = 500; const MAX_BACKOFF_MS = 15_000; const POLL_INTERVAL_MS = 3_000; +/** Refresh this far ahead of expiry, so a request in flight never crosses the boundary. */ +const EXPIRY_MARGIN_MS = 60_000; export class SyncServiceClient { private socket?: WebSocket; @@ -53,11 +56,19 @@ export class SyncServiceClient { private handlers?: SyncStreamHandlers; private cursorProvider: () => number = () => 0; + private session?: StoredSession; + private refreshing?: Promise; + constructor( private readonly config: SyncDaemonConfig, readonly branch: string, - private replicaId?: string - ) {} + private replicaId?: string, + /** Called with a refreshed session so the caller can persist it for the next start-up. */ + private readonly onSession?: (session: StoredSession) => Promise, + private readonly pollIntervalMs = POLL_INTERVAL_MS + ) { + this.session = config.service.session; + } /** In touch with the service, by either transport -- what `crosscode status` reports. */ get connected(): boolean { @@ -111,7 +122,7 @@ export class SyncServiceClient { this.handlers = handlers; this.cursorProvider = cursor; this.stopped = false; - this.connect(); + void this.connect(); } stop(): void { @@ -124,9 +135,14 @@ export class SyncServiceClient { socket?.close(); } - private connect(): void { + private async connect(): Promise { if (this.stopped || !this.replicaId) return; - const socket = new WebSocket(`${streamUrl(this.config.service.url)}/v1/stream`, this.authorization() ? { headers: { authorization: this.authorization()! } } : {}); + // A refresh can fail -- a revoked or rotated-away refresh token -- and that must not + // take the daemon down from a reconnect timer. Falling through to polling is the + // honest outcome: it fails the same way, and reports `connected: false`. + const authorization = await this.authorization().catch(() => undefined); + if (this.stopped) return; + const socket = new WebSocket(`${streamUrl(this.config.service.url)}/v1/stream`, authorization ? { headers: { authorization } } : {}); this.socket = socket; socket.on("open", () => { socket.send(JSON.stringify({ @@ -164,7 +180,7 @@ export class SyncServiceClient { if (this.stopped) return; const delay = this.backoffMs; this.backoffMs = Math.min(this.backoffMs * 2, MAX_BACKOFF_MS); - this.reconnectTimer = setTimeout(() => this.connect(), delay); + this.reconnectTimer = setTimeout(() => void this.connect(), delay); this.reconnectTimer.unref(); } @@ -188,7 +204,7 @@ export class SyncServiceClient { this.polling = false; } if (this.stopped || this.streaming) { this.stopPolling(); return; } - this.pollTimer = setTimeout(() => void tick(), POLL_INTERVAL_MS); + this.pollTimer = setTimeout(() => void tick(), this.pollIntervalMs); this.pollTimer.unref(); }; void tick(); @@ -206,13 +222,31 @@ export class SyncServiceClient { return this.replicaId; } - private authorization(): string | undefined { - const token = this.config.service.session?.accessToken; - return token ? `Bearer ${token}` : undefined; + /** + * The bearer header, refreshed if it is about to expire. Supabase access tokens last an + * hour; a daemon is expected to outlive that, so every authenticated call goes through + * here rather than reading the token that was written at sign-in. + * + * Concurrent callers share one refresh via `refreshing`: the socket reconnecting and a + * publish landing at the same moment would otherwise each spend the refresh token, and + * Supabase rotates it, so the second would be spending one that no longer exists. + */ + private async authorization(): Promise { + const session = this.session; + if (!session) return undefined; + if (Date.parse(session.expiresAt) - Date.now() > EXPIRY_MARGIN_MS) return `Bearer ${session.accessToken}`; + this.refreshing ??= refreshStoredSession(session) + .then(async (refreshed) => { + this.session = refreshed; + await this.onSession?.(refreshed); + return refreshed; + }) + .finally(() => { this.refreshing = undefined; }); + return `Bearer ${(await this.refreshing).accessToken}`; } private async request(method: "GET" | "POST", path: string, body?: unknown): Promise { - const authorization = this.authorization(); + const authorization = await this.authorization(); const response = await fetch(new URL(path, this.config.service.url), { method, headers: { From 663f9ddfd7c18ac3d463886726d610ba8a025200 Mon Sep 17 00:00:00 2001 From: Abdullah Sultan Date: Thu, 6 Aug 2026 13:09:31 -0500 Subject: [PATCH 3/6] Say why the sync stopped, instead of going quiet A failed poll set `connected: false` and discarded the reason. The daemon then sat there with a cursor that never advanced and nothing on stderr -- which is exactly the state a production outage put it in, and the reason diagnosing it needed a patched build rather than a log line. Reported once per run of failures rather than every 3s, and reset when a poll succeeds, so a flapping service does not fill the terminal. This is what surfaced `GET /v1/changes` answering 500 with `relation "terms_acceptances" does not exist` -- a migration that shipped in the deploy but was never applied. Co-Authored-By: Claude Opus 5 --- apps/daemon/src/sync-service-client.ts | 11 ++++++++++- 1 file changed, 10 insertions(+), 1 deletion(-) diff --git a/apps/daemon/src/sync-service-client.ts b/apps/daemon/src/sync-service-client.ts index d3569bb..d8e0c51 100644 --- a/apps/daemon/src/sync-service-client.ts +++ b/apps/daemon/src/sync-service-client.ts @@ -51,6 +51,7 @@ export class SyncServiceClient { private pollTimer?: NodeJS.Timeout; private pollActive = false; private polling = false; + private reportedPollFailure = false; private backoffMs = INITIAL_BACKOFF_MS; private stopped = false; private handlers?: SyncStreamHandlers; @@ -199,8 +200,16 @@ export class SyncServiceClient { try { await this.handlers?.onPoll?.(); this.polling = true; - } catch { + this.reportedPollFailure = false; + } catch (error) { // Unreachable by either transport, which is what `connected: false` should mean. + // Reported once per run of failures rather than every 3s: a cursor that stops + // advancing is otherwise completely silent, and `connected: false` on its own does + // not say whether the service is down or a change cannot be applied. + if (!this.reportedPollFailure) { + this.reportedPollFailure = true; + process.stderr.write(`Crosscode: sync poll failed: ${error instanceof Error ? error.message : String(error)}\n`); + } this.polling = false; } if (this.stopped || this.streaming) { this.stopPolling(); return; } From b9d7a638f50c8d2f2fc70b8007d0c941cd4fc1f3 Mon Sep 17 00:00:00 2001 From: Abdullah Sultan Date: Thu, 6 Aug 2026 13:21:55 -0500 Subject: [PATCH 4/6] Stop one unrebasable change from spinning the service forever apply() calls catchUp() to fill a gap when a peer's change is built on a blob this checkout has never held. Called from inside a drain, catchUp re-walked the same page from the same cursor -- and the cursor cannot move past a change that has not applied, so it never moved. The retry was already guarded: allowCatchup: false on the second receive exists so a second miss surfaces rather than loops. But the fill in front of it was not, and that is where the recursion was. In production this was one peer asking GET /v1/changes several times a second, indefinitely, while crosscode status showed a cursor frozen at 1 and reported no error at all. A fresh checkout that joined a project it could not fully rebase never received anything again. catchUp is now non-re-entrant: a nested call is a no-op, because the drain already running is the fill. The retry then decides, and an honest conflict reaches the agent. The stub counts GET /v1/changes, and the test asserts a daemon holding a change whose baseHash names a blob that exists nowhere still advances its cursor and then goes quiet. Verified it fails without the guard: the cursor never advances and it times out. Co-Authored-By: Claude Opus 5 --- apps/daemon/src/sync-daemon.test.ts | 34 ++++++++++++++++++++++++++++ apps/daemon/src/sync-daemon.ts | 20 ++++++++++++++++ apps/daemon/src/sync-service-stub.ts | 6 ++++- 3 files changed, 59 insertions(+), 1 deletion(-) diff --git a/apps/daemon/src/sync-daemon.test.ts b/apps/daemon/src/sync-daemon.test.ts index 5fb96e9..73d1236 100644 --- a/apps/daemon/src/sync-daemon.test.ts +++ b/apps/daemon/src/sync-daemon.test.ts @@ -280,6 +280,40 @@ describe("the daemon around the engine", () => { expect(descriptor).toBeNull(); }, 60_000); + /** + * A change built on a blob this checkout has never held makes apply() call catchUp() to + * fill the gap. Called from inside a drain, that re-walked the same page from the same + * cursor -- and the cursor cannot move past a change that has not applied, so it never + * moved. One such change from one peer span the real service at several requests a second + * indefinitely, while `crosscode status` showed `cursor` frozen and no error anywhere. + */ + it("does not spin the service on a change it cannot rebase", async () => { + const service = await startSyncServiceStub(); + services.push(service); + const origin = await seedOrigin({ "a.txt": "0\n" }); + const root = await checkout(origin, "alice", service); + // A modify whose baseHash names a blob that exists nowhere: the sender's history is one + // this checkout has no way to reach. + service.changes.push({ + sequence: 1, projectId: "project-1", branch: "main", + replicaId: "00000000-1111-4222-8333-444455556666", + createdAt: "2026-01-01T00:00:00.000Z", + version: { + path: "a.txt", op: "modify", baseHash: "0".repeat(40), + contentHash: "1".repeat(40), content: "from a history we never had\n", encoding: "utf8" + } + }); + + const daemon = await start(root); + await waitFor("the cursor to get past it", async () => daemon.status().cursor >= 1); + const afterSettling = service.listCalls; + await new Promise((settle) => setTimeout(settle, 1_000)); + + // Ten polls a second would be a spin; the configured interval is 100ms in these tests, + // so a healthy daemon asks a handful of times in a second and no more. + expect(service.listCalls - afterSettling).toBeLessThan(20); + }, 60_000); + it("resyncs from full content when the service says the cursor is too old", async () => { const service = await startSyncServiceStub(); services.push(service); diff --git a/apps/daemon/src/sync-daemon.ts b/apps/daemon/src/sync-daemon.ts index 18c95ac..67086c4 100644 --- a/apps/daemon/src/sync-daemon.ts +++ b/apps/daemon/src/sync-daemon.ts @@ -70,6 +70,7 @@ class HttpError extends Error { export class SyncDaemon { private watcher?: FSWatcher; private server?: Server; + private draining = false; private descriptor?: DaemonDescriptor; /** Built when the loopback server binds, published only once start-up has fully succeeded. */ private pendingDescriptor?: DaemonDescriptor; @@ -259,6 +260,13 @@ export class SyncDaemon { // The sender built on a blob we have never held, so we are genuinely behind. Fill // the gap from the cursor and retry once; a second miss has to surface rather than // loop, which is what allowCatchup: false forces. + // + // That guard covers the retry but not the fill: called from inside a drain, catchUp + // re-walks the same page from the same cursor -- which the cursor cannot leave, because + // it only advances past a change that applied. One unrebasable change from a peer + // therefore span the service at several requests a second, forever, while `crosscode + // status` showed a cursor that never moved. The guard inside catchUp makes the nested + // call a no-op, so the retry below decides, and an honest conflict reaches the agent. await this.catchUp(); result = await this.engine.receive(change.version, { peer: change.replicaId, allowCatchup: false }); } @@ -279,6 +287,18 @@ export class SyncDaemon { * is what turns a dropped socket into a gap that closes itself. */ private async catchUp(): Promise { + // Not re-entrant: apply() calls this to fill a gap, and the drain that is already + // running is the fill. See the note there for what the recursion cost. + if (this.draining) return; + this.draining = true; + try { + await this.drainChanges(); + } finally { + this.draining = false; + } + } + + private async drainChanges(): Promise { for (let page = 0; page < 100; page += 1) { const response = await this.client.changes(this.cursor); if ("status" in response) { diff --git a/apps/daemon/src/sync-service-stub.ts b/apps/daemon/src/sync-service-stub.ts index d7b59f9..44e7850 100644 --- a/apps/daemon/src/sync-service-stub.ts +++ b/apps/daemon/src/sync-service-stub.ts @@ -28,6 +28,8 @@ type Subscriber = { socket: WebSocket; projectId: string; branch: string; replic export type SyncServiceStub = { url: string; changes: Change[]; + /** How many times `GET /v1/changes` has been asked, so a test can catch a spin. */ + readonly listCalls: number; /** Sequences at or below this are "no longer retained", to exercise cursor-too-old. */ retentionFloor: number; /** Drops every open stream, as a network partition would. */ @@ -48,7 +50,7 @@ export async function startSyncServiceStub(options: SyncServiceStubOptions = {}) const changes: Change[] = []; const subscribers = new Set(); let sequence = 0; - const state = { retentionFloor: 0 }; + const state = { retentionFloor: 0, listCalls: 0 }; const send = (response: import("node:http").ServerResponse, status: number, body: unknown) => { response.writeHead(status, { "content-type": "application/json" }); @@ -87,6 +89,7 @@ export async function startSyncServiceStub(options: SyncServiceStubOptions = {}) return; } if (request.method === "GET" && url.pathname === "/v1/changes") { + state.listCalls += 1; const query = listChangesQuerySchema.parse({ projectId: url.searchParams.get("projectId"), branch: url.searchParams.get("branch"), @@ -156,6 +159,7 @@ export async function startSyncServiceStub(options: SyncServiceStubOptions = {}) return { url: `http://127.0.0.1:${port}`, changes, + get listCalls() { return state.listCalls; }, get retentionFloor() { return state.retentionFloor; }, set retentionFloor(value: number) { state.retentionFloor = value; }, disconnectAll() { From 7d98e314e1d77bda977ab9794cb914153ad3c7df Mon Sep 17 00:00:00 2001 From: Abdullah Sultan Date: Thu, 6 Aug 2026 13:32:58 -0500 Subject: [PATCH 5/6] Give each direction its own file in the polling test Bouncing one file straight back landed the reply inside the hot window the first write opened, so under CI's load it was a deferral test wearing a transport test's name -- and it failed there while passing locally in under two seconds. Co-Authored-By: Claude Opus 5 --- apps/daemon/src/sync-daemon.test.ts | 9 ++++++--- 1 file changed, 6 insertions(+), 3 deletions(-) diff --git a/apps/daemon/src/sync-daemon.test.ts b/apps/daemon/src/sync-daemon.test.ts index 73d1236..7f33792 100644 --- a/apps/daemon/src/sync-daemon.test.ts +++ b/apps/daemon/src/sync-daemon.test.ts @@ -230,19 +230,22 @@ describe("the daemon around the engine", () => { it("still receives a peer's edits when the host cannot serve a stream at all", async () => { const service = await startSyncServiceStub({ streams: false }); services.push(service); - const origin = await seedOrigin({ "a.txt": "0\n" }); + const origin = await seedOrigin({ "a.txt": "0\n", "b.txt": "0\n" }); const roots = await Promise.all(["alice", "bob"].map((name) => checkout(origin, name, service))); const [alice, bob] = [await start(roots[0]!), await start(roots[1]!)]; // Reachable by the only transport there is, and honest about saying so. await waitFor("both daemons to report themselves connected", async () => alice.status().connected && bob.status().connected); + // Each direction gets its own file. Bouncing one file back immediately would land the + // reply inside the hot window the first write opened, which is a deferral test, not a + // transport one. await type(roots[0]!, "a.txt", "arrived without a socket\n"); await waitFor("bob to receive it", () => allEqual(roots, "a.txt", "arrived without a socket\n")); // And the other way, so it is a two-way sync rather than one lucky direction. - await type(roots[1]!, "a.txt", "and back again\n"); - await waitFor("alice to receive it", () => allEqual(roots, "a.txt", "and back again\n")); + await type(roots[1]!, "b.txt", "and back again\n"); + await waitFor("alice to receive it", () => allEqual(roots, "b.txt", "and back again\n")); }, 60_000); /** From 3634abd45093bf2371e11ca267263b99d3278d85 Mon Sep 17 00:00:00 2001 From: Abdullah Sultan Date: Thu, 6 Aug 2026 13:49:13 -0500 Subject: [PATCH 6/6] Stop syncing the files Crosscode installs into the checkout The first time two real checkouts joined one project, the teammate arrived holding three conflicts before typing anything: .mcp.json, .claude/settings.json, and the crosscode SKILL.md. Every checkout installs its own copy, so no two peers share an ancestor for them. They could never converge either, because `crosscode start` rewrites them locally on every run -- so any resolution is undone by the next start. And they are not content in the sense the rest of the sync means: they are per-machine setup, regenerated on demand by the command the invite already tells a teammate to run. Nothing is lost by leaving them out, and nothing was fixable by leaving them in. Added to isDenied, which is already the one choke point checked on the way out and on the way in. The rest of .claude/ still syncs -- a team's own skills and commands are theirs to share, and only the three paths Crosscode itself writes are excluded. Verified live: the two checkouts that were holding these three conflicts came back up with zero, and still sync in both directions. Co-Authored-By: Claude Opus 5 --- packages/sync/src/paths.test.ts | 31 +++++++++++++++++++++++++++++++ packages/sync/src/paths.ts | 21 ++++++++++++++++++++- 2 files changed, 51 insertions(+), 1 deletion(-) create mode 100644 packages/sync/src/paths.test.ts diff --git a/packages/sync/src/paths.test.ts b/packages/sync/src/paths.test.ts new file mode 100644 index 0000000..a140a27 --- /dev/null +++ b/packages/sync/src/paths.test.ts @@ -0,0 +1,31 @@ +import { describe, expect, it } from "vitest"; +import { isDenied, isSafeRelativePath } from "./paths.js"; + +describe("what Crosscode refuses to sync", () => { + it("refuses the files `crosscode start` installs into the checkout", () => { + // Observed the first time two real checkouts joined one project: the teammate arrived + // holding three conflicts before typing anything, because each checkout had installed + // its own copy of these and no two shared an ancestor. + expect(isDenied(".mcp.json")).toBe(true); + expect(isDenied(".claude/settings.json")).toBe(true); + expect(isDenied(".claude/skills/crosscode/SKILL.md")).toBe(true); + }); + + it("still syncs the rest of the agent's directory, which is the team's to share", () => { + expect(isDenied(".claude/skills/deploy/SKILL.md")).toBe(false); + expect(isDenied(".claude/commands/review.md")).toBe(false); + expect(isDenied("src/index.ts")).toBe(false); + }); + + it("still refuses secrets, which is what the denylist was for first", () => { + expect(isDenied(".env")).toBe(true); + expect(isDenied(".env.local")).toBe(true); + }); + + it("refuses anything that would escape the checkout or touch git's own state", () => { + expect(isSafeRelativePath("../outside")).toBe(false); + expect(isSafeRelativePath("/etc/passwd")).toBe(false); + expect(isSafeRelativePath(".git/config")).toBe(false); + expect(isSafeRelativePath("src/a.ts")).toBe(true); + }); +}); diff --git a/packages/sync/src/paths.ts b/packages/sync/src/paths.ts index 1ceb3a8..f8d4626 100644 --- a/packages/sync/src/paths.ts +++ b/packages/sync/src/paths.ts @@ -18,5 +18,24 @@ export function isSafeRelativePath(path: string): boolean { * those would be a serious incident, so it is checked on the way out *and* on the way in. */ export function isDenied(path: string): boolean { - return redactPath(path); + return redactPath(path) || isCrosscodeOwned(path); +} + +/** + * The files `crosscode start` writes into the checkout itself: the MCP registration, the + * skill, and the hook in the agent's settings. + * + * Every checkout installs its own copy, so no two peers share an ancestor for them, and + * syncing them means the first teammate to join arrives holding a conflict on all three + * before they have typed anything. Worse, they can never converge: `start` rewrites them + * locally on every run, so a resolution is undone by the next start. + * + * They are also not content in the sense the rest of the sync means. They are per-machine + * setup, regenerated on demand by the command the invite already tells a teammate to run. + * Nothing is lost by leaving them out, and nothing is fixable by leaving them in. + */ +function isCrosscodeOwned(path: string): boolean { + return path === ".mcp.json" + || path === ".claude/settings.json" + || path.startsWith(".claude/skills/crosscode/"); }