Load thread snapshots over HTTP before live sync#3719
Open
juliusmarminge wants to merge 5 commits into
Open
Conversation
- Fetch initial thread snapshots via HTTP and resume subscriptions after the snapshot sequence - Add not-found handling for thread snapshot requests
|
Important Review skippedAuto reviews are disabled on this repository. Please check the settings in the CodeRabbit UI or the ⚙️ Run configurationConfiguration used: Repository UI Review profile: CHILL Plan: Pro Run ID: You can disable this status message by setting the Use the checkbox below for a quick retry:
✨ Finishing Touches🧪 Generate unit tests (beta)
Comment |
github-actions
Bot
added
vouch:trusted
![macroscopeapp[bot]](https://avatars.githubusercontent.com/in/900172?s=60&v=4){kind=small}
| retryExpectedFailureAfter: "250 millis", | ||
| }, | ||
| ).pipe(Stream.runForEach(applyItem), Effect.forkScoped); | ||
| yield* Effect.forkScoped( |
Contributor
There was a problem hiding this comment.
🟡 Medium state/threads.ts:193
makeEnvironmentThreadState now awaits snapshotLoader.load(...) before calling subscribe(...), so when the HTTP snapshot endpoint is slow or times out, live WebSocket updates are blocked for the entire duration even though the socket session is already healthy. The UI stays in stale cached/synchronizing state and misses live updates until the HTTP fallback completes. Consider subscribing to live events immediately and applying the HTTP snapshot in parallel when it arrives — applyItem already deduplicates by sequence, so overlapping events are safe.
🚀 Reply "fix it for me" or copy this AI Prompt for your agent:
In file @packages/client-runtime/src/state/threads.ts around line 193:
`makeEnvironmentThreadState` now awaits `snapshotLoader.load(...)` before calling `subscribe(...)`, so when the HTTP snapshot endpoint is slow or times out, live WebSocket updates are blocked for the entire duration even though the socket session is already healthy. The UI stays in stale `cached`/`synchronizing` state and misses live updates until the HTTP fallback completes. Consider subscribing to live events immediately and applying the HTTP snapshot in parallel when it arrives — `applyItem` already deduplicates by sequence, so overlapping events are safe.
Member
Author
There was a problem hiding this comment.
Mitigated in c420e9e. The snapshot fetch requires the sequence before we can resume live events over the socket, so some coupling is inherent, but: the cached thread renders while the fetch runs (so first paint is not blocked), the wait is bounded (timeout lowered to 6s with fallback to the socket snapshot), and this only affects the initial mount — reconnects resume via the socket without re-fetching.
![cursor[bot]](https://avatars.githubusercontent.com/in/1210556?s=60&v=4){kind=small}
Contributor
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using high effort and found 3 potential issues.
Autofix Details
Bugbot Autofix prepared fixes for all 3 issues found in the latest run.
- ✅ Fixed: PubSub attached after replay
- Added OrchestrationEngine.subscribeDomainEvents, which attaches the PubSub subscription (scoped to the RPC stream's lifetime) before subscribeThread reads its catch-up or snapshot baseline, so events published during the baseline read are buffered instead of dropped.
- ✅ Fixed: Thread catch-up hits replay cap
- readEvents now accepts an optional limit forwarded to the event store, and the afterSequence catch-up passes Number.MAX_SAFE_INTEGER so the replay is exhaustive (internally paged) instead of truncating at the global 1,000-event default.
- ✅ Fixed: HTTP snapshot reads race
- Added a transactional ProjectionSnapshotQuery.getThreadDetailSnapshot that reads the thread detail and snapshot sequence in one sql.withTransaction, now used by both the HTTP threadSnapshot endpoint and the WS snapshot path so the sequence never runs ahead of the embedded thread.
Or push these changes by commenting:
@cursor push 29677d1179
Preview (29677d1179)
diff --git a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts
--- a/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts
+++ b/apps/server/src/checkpointing/CheckpointDiffQuery.test.ts
@@ -107,6 +107,7 @@
}),
getThreadShellById: () => Effect.succeed(Option.none()),
getThreadDetailById: () => Effect.succeed(Option.none()),
+ getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
}),
),
);
@@ -199,6 +200,7 @@
getFullThreadDiffContext: () => Effect.die("unused"),
getThreadShellById: () => Effect.succeed(Option.none()),
getThreadDetailById: () => Effect.succeed(Option.none()),
+ getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
}),
),
);
@@ -281,6 +283,7 @@
getFullThreadDiffContext: () => Effect.die("unused"),
getThreadShellById: () => Effect.succeed(Option.none()),
getThreadDetailById: () => Effect.succeed(Option.none()),
+ getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
}),
),
);
@@ -348,6 +351,7 @@
getFullThreadDiffContext: () => Effect.die("unused"),
getThreadShellById: () => Effect.succeed(Option.none()),
getThreadDetailById: () => Effect.succeed(Option.none()),
+ getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
}),
),
);
@@ -400,6 +404,7 @@
getFullThreadDiffContext: () => Effect.succeed(Option.none()),
getThreadShellById: () => Effect.succeed(Option.none()),
getThreadDetailById: () => Effect.succeed(Option.none()),
+ getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
}),
),
);
diff --git a/apps/server/src/observability/RpcInstrumentation.ts b/apps/server/src/observability/RpcInstrumentation.ts
--- a/apps/server/src/observability/RpcInstrumentation.ts
+++ b/apps/server/src/observability/RpcInstrumentation.ts
@@ -5,6 +5,7 @@
import * as Exit from "effect/Exit";
import * as Metric from "effect/Metric";
import * as References from "effect/References";
+import type * as Scope from "effect/Scope";
import * as Stream from "effect/Stream";
import { outcomeFromExit } from "./Attributes.ts";
@@ -123,7 +124,14 @@
method: string,
effect: Effect.Effect<Stream.Stream<A, StreamError, StreamContext>, EffectError, EffectContext>,
traceAttributes?: Readonly<Record<string, unknown>>,
-): Stream.Stream<A, StreamError | EffectError, StreamContext | EffectContext> => {
+ // `Stream.unwrap` scopes the setup effect to the stream's lifetime, so a
+ // `Scope` requirement (e.g. PubSub subscriptions attached before a snapshot
+ // is loaded) is satisfied by the stream itself rather than the caller.
+): Stream.Stream<
+ A,
+ StreamError | EffectError,
+ StreamContext | Exclude<EffectContext, Scope.Scope>
+> => {
const instrumented = Stream.unwrap(
Effect.gen(function* () {
const startedAt = yield* Clock.currentTimeNanos;
diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts
--- a/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts
+++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.test.ts
@@ -200,6 +200,7 @@
getFullThreadDiffContext: () => Effect.succeed(Option.none()),
getThreadShellById: () => Effect.succeed(Option.none()),
getThreadDetailById: () => Effect.succeed(Option.none()),
+ getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
}),
),
Layer.provide(
diff --git a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts
--- a/apps/server/src/orchestration/Layers/OrchestrationEngine.ts
+++ b/apps/server/src/orchestration/Layers/OrchestrationEngine.ts
@@ -306,8 +306,8 @@
Effect.annotateLogs({ sequence: commandReadModel.snapshotSequence }),
);
- const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive) =>
- eventStore.readFromSequence(fromSequenceExclusive);
+ const readEvents: OrchestrationEngineShape["readEvents"] = (fromSequenceExclusive, limit) =>
+ eventStore.readFromSequence(fromSequenceExclusive, limit);
const dispatch: OrchestrationEngineShape["dispatch"] = (command) =>
Effect.gen(function* () {
@@ -329,6 +329,7 @@
get streamDomainEvents(): OrchestrationEngineShape["streamDomainEvents"] {
return Stream.fromPubSub(eventPubSub);
},
+ subscribeDomainEvents: Effect.map(PubSub.subscribe(eventPubSub), Stream.fromSubscription),
} satisfies OrchestrationEngineShape;
});
diff --git a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts
--- a/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts
+++ b/apps/server/src/orchestration/Layers/ProjectionSnapshotQuery.ts
@@ -2033,6 +2033,23 @@
);
});
+ const getThreadDetailSnapshot: ProjectionSnapshotQueryShape["getThreadDetailSnapshot"] = (
+ threadId,
+ ) =>
+ sql.withTransaction(Effect.all([getThreadDetailById(threadId), getSnapshotSequence()])).pipe(
+ Effect.map(([threadDetail, { snapshotSequence }]) =>
+ Option.map(threadDetail, (thread) => ({ snapshotSequence, thread })),
+ ),
+ Effect.mapError((error) => {
+ if (isPersistenceError(error)) {
+ return error;
+ }
+ return toPersistenceSqlError("ProjectionSnapshotQuery.getThreadDetailSnapshot:query")(
+ error,
+ );
+ }),
+ );
+
return {
getCommandReadModel,
getSnapshot,
@@ -2047,6 +2064,7 @@
getFullThreadDiffContext,
getThreadShellById,
getThreadDetailById,
+ getThreadDetailSnapshot,
} satisfies ProjectionSnapshotQueryShape;
});
diff --git a/apps/server/src/orchestration/Services/OrchestrationEngine.ts b/apps/server/src/orchestration/Services/OrchestrationEngine.ts
--- a/apps/server/src/orchestration/Services/OrchestrationEngine.ts
+++ b/apps/server/src/orchestration/Services/OrchestrationEngine.ts
@@ -13,6 +13,7 @@
import type { OrchestrationCommand, OrchestrationEvent } from "@t3tools/contracts";
import * as Context from "effect/Context";
import type * as Effect from "effect/Effect";
+import type * as Scope from "effect/Scope";
import type * as Stream from "effect/Stream";
import type { OrchestrationDispatchError } from "../Errors.ts";
@@ -26,10 +27,14 @@
* Replay persisted orchestration events from an exclusive sequence cursor.
*
* @param fromSequenceExclusive - Sequence cursor (exclusive).
+ * @param limit - Optional maximum number of events to replay. Defaults to
+ * the event store's replay cap; pass `Number.MAX_SAFE_INTEGER` for an
+ * exhaustive catch-up replay.
* @returns Stream containing ordered events.
*/
readonly readEvents: (
fromSequenceExclusive: number,
+ limit?: number,
) => Stream.Stream<OrchestrationEvent, OrchestrationEventStoreError, never>;
/**
@@ -49,8 +54,26 @@
* Stream persisted domain events in dispatch order.
*
* This is a hot runtime stream (new events only), not a historical replay.
+ * The underlying PubSub subscription is only attached once the stream is
+ * pulled; use `subscribeDomainEvents` when the subscription must be
+ * attached before other work (e.g. loading a snapshot baseline).
*/
readonly streamDomainEvents: Stream.Stream<OrchestrationEvent>;
+
+ /**
+ * Attach a domain event subscription immediately and return the stream of
+ * events it receives.
+ *
+ * Unlike `streamDomainEvents`, events published between running this effect
+ * and pulling the returned stream are buffered by the subscription instead
+ * of dropped, which is required for snapshot/catch-up + live combinations.
+ * The subscription is released when the surrounding scope closes.
+ */
+ readonly subscribeDomainEvents: Effect.Effect<
+ Stream.Stream<OrchestrationEvent>,
+ never,
+ Scope.Scope
+ >;
}
/**
diff --git a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts
--- a/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts
+++ b/apps/server/src/orchestration/Services/ProjectionSnapshotQuery.ts
@@ -14,6 +14,7 @@
OrchestrationReadModel,
OrchestrationShellSnapshot,
OrchestrationThread,
+ OrchestrationThreadDetailSnapshot,
OrchestrationThreadShell,
ProjectId,
ThreadId,
@@ -157,6 +158,15 @@
readonly getThreadDetailById: (
threadId: ThreadId,
) => Effect.Effect<Option.Option<OrchestrationThread>, ProjectionRepositoryError>;
+
+ /**
+ * Read a single active thread detail together with the projection snapshot
+ * sequence, both observed inside one transaction so the sequence never runs
+ * ahead of the thread rows it is paired with.
+ */
+ readonly getThreadDetailSnapshot: (
+ threadId: ThreadId,
+ ) => Effect.Effect<Option.Option<OrchestrationThreadDetailSnapshot>, ProjectionRepositoryError>;
}
/**
diff --git a/apps/server/src/orchestration/http.ts b/apps/server/src/orchestration/http.ts
--- a/apps/server/src/orchestration/http.ts
+++ b/apps/server/src/orchestration/http.ts
@@ -45,18 +45,17 @@
Effect.fn("environment.orchestration.threadSnapshot")(function* (args) {
yield* annotateEnvironmentRequest(args.endpoint.name);
yield* requireEnvironmentScope(AuthOrchestrationReadScope);
- const [threadDetail, { snapshotSequence }] = yield* Effect.all([
- projectionSnapshotQuery.getThreadDetailById(args.params.threadId),
- projectionSnapshotQuery.getSnapshotSequence(),
- ]).pipe(
- Effect.catch((cause) =>
- failEnvironmentInternal("orchestration_thread_snapshot_failed", cause),
- ),
- );
- if (Option.isNone(threadDetail)) {
+ const snapshot = yield* projectionSnapshotQuery
+ .getThreadDetailSnapshot(args.params.threadId)
+ .pipe(
+ Effect.catch((cause) =>
+ failEnvironmentInternal("orchestration_thread_snapshot_failed", cause),
+ ),
+ );
+ if (Option.isNone(snapshot)) {
return yield* failEnvironmentNotFound("thread_not_found");
}
- return { snapshotSequence, thread: threadDetail.value };
+ return snapshot.value;
}),
)
.handle(
diff --git a/apps/server/src/project/ProjectSetupScriptRunner.test.ts b/apps/server/src/project/ProjectSetupScriptRunner.test.ts
--- a/apps/server/src/project/ProjectSetupScriptRunner.test.ts
+++ b/apps/server/src/project/ProjectSetupScriptRunner.test.ts
@@ -43,6 +43,7 @@
getFullThreadDiffContext: () => Effect.die("unused"),
getThreadShellById: () => Effect.die("unused"),
getThreadDetailById: () => Effect.die("unused"),
+ getThreadDetailSnapshot: () => Effect.die("unused"),
});
const makeTerminalManagerLayer = (
diff --git a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts
--- a/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts
+++ b/apps/server/src/provider/Layers/ProviderSessionReaper.test.ts
@@ -209,6 +209,7 @@
: Option.none(),
),
getThreadDetailById: () => Effect.die("unused"),
+ getThreadDetailSnapshot: () => Effect.die("unused"),
}),
),
Layer.provideMerge(NodeServices.layer),
diff --git a/apps/server/src/relay/AgentAwarenessRelay.test.ts b/apps/server/src/relay/AgentAwarenessRelay.test.ts
--- a/apps/server/src/relay/AgentAwarenessRelay.test.ts
+++ b/apps/server/src/relay/AgentAwarenessRelay.test.ts
@@ -461,6 +461,7 @@
readEvents: () => Stream.empty,
dispatch: () => Effect.succeed({ sequence: 1 }),
streamDomainEvents: Stream.fromQueue(events),
+ subscribeDomainEvents: Effect.sync(() => Stream.fromQueue(events)),
} satisfies OrchestrationEngineShape;
const snapshotQuery = {
@@ -650,6 +651,7 @@
readEvents: () => Stream.empty,
dispatch: () => Effect.succeed({ sequence: 1 }),
streamDomainEvents: Stream.fromQueue(events),
+ subscribeDomainEvents: Effect.sync(() => Stream.fromQueue(events)),
} satisfies OrchestrationEngineShape),
Layer.succeed(ProjectionSnapshotQuery, {
getShellSnapshot: () =>
diff --git a/apps/server/src/serverRuntimeStartup.test.ts b/apps/server/src/serverRuntimeStartup.test.ts
--- a/apps/server/src/serverRuntimeStartup.test.ts
+++ b/apps/server/src/serverRuntimeStartup.test.ts
@@ -96,6 +96,7 @@
getFullThreadDiffContext: () => Effect.succeed(Option.none()),
getThreadShellById: () => Effect.succeed(Option.none()),
getThreadDetailById: () => Effect.succeed(Option.none()),
+ getThreadDetailSnapshot: () => Effect.succeed(Option.none()),
}),
Effect.provideService(AnalyticsService.AnalyticsService, {
record: () => Effect.void,
@@ -158,6 +159,7 @@
getFullThreadDiffContext: () => Effect.succeed(Option.none()),
getThreadShellById: () => Effect.die("unused"),
getThreadDetailById: () => Effect.die("unused"),
+ getThreadDetailSnapshot: () => Effect.die("unused"),
}),
Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
readEvents: () => Stream.empty,
@@ -166,6 +168,7 @@
Effect.as({ sequence: 1 }),
),
streamDomainEvents: Stream.empty,
+ subscribeDomainEvents: Effect.succeed(Stream.empty),
} satisfies OrchestrationEngine.OrchestrationEngineService["Service"]),
Effect.provide(NodeServices.layer),
);
@@ -200,6 +203,7 @@
getFullThreadDiffContext: () => Effect.succeed(Option.none()),
getThreadShellById: () => Effect.die("unused"),
getThreadDetailById: () => Effect.die("unused"),
+ getThreadDetailSnapshot: () => Effect.die("unused"),
}),
Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
readEvents: () => Stream.empty,
@@ -208,6 +212,7 @@
Effect.as({ sequence: 1 }),
),
streamDomainEvents: Stream.empty,
+ subscribeDomainEvents: Effect.succeed(Stream.empty),
} satisfies OrchestrationEngine.OrchestrationEngineService["Service"]),
Effect.provide(NodeServices.layer),
);
@@ -248,6 +253,7 @@
getFullThreadDiffContext: () => Effect.succeed(Option.none()),
getThreadShellById: () => Effect.die("unused"),
getThreadDetailById: () => Effect.die("unused"),
+ getThreadDetailSnapshot: () => Effect.die("unused"),
}),
Effect.provideService(OrchestrationEngine.OrchestrationEngineService, {
readEvents: () => Stream.empty,
@@ -256,6 +262,7 @@
Effect.as({ sequence: 1 }),
),
streamDomainEvents: Stream.empty,
+ subscribeDomainEvents: Effect.succeed(Stream.empty),
} satisfies OrchestrationEngine.OrchestrationEngineService["Service"]),
Effect.provideService(Crypto.Crypto, {
...crypto,
diff --git a/apps/server/src/ws.ts b/apps/server/src/ws.ts
--- a/apps/server/src/ws.ts
+++ b/apps/server/src/ws.ts
@@ -1119,7 +1119,12 @@
event.aggregateId === input.threadId &&
isThreadDetailEvent(event);
- const liveStream = orchestrationEngine.streamDomainEvents.pipe(
+ // Attach the domain event subscription before loading the
+ // snapshot/catch-up baseline: the PubSub does not buffer for
+ // late subscribers, so events published while the baseline is
+ // read would otherwise be dropped. Overlapping events are
+ // deduped by sequence on the client.
+ const liveStream = (yield* orchestrationEngine.subscribeDomainEvents).pipe(
Stream.filter(isThisThreadDetailEvent),
Stream.map((event) => ({
kind: "event" as const,
@@ -1131,26 +1136,29 @@
// that snapshot's sequence, and we resume the live subscription by
// replaying persisted events after it instead of re-sending the
// (potentially multi-KB) snapshot frame over the socket. The
- // catch-up replay + live stream carry the same ordering guarantees
- // as the snapshot-then-live path below; overlapping events are
- // deduped by sequence on the client.
+ // catch-up replay must be exhaustive — a truncated replay would
+ // silently drop thread updates — so it bypasses the default
+ // replay cap.
if (input.afterSequence !== undefined) {
- const catchUpStream = orchestrationEngine.readEvents(input.afterSequence).pipe(
- Stream.filter(isThisThreadDetailEvent),
- Stream.map((event) => ({ kind: "event" as const, event })),
- Stream.mapError(
- (cause) =>
- new OrchestrationGetSnapshotError({
- message: `Failed to replay thread ${input.threadId} events`,
- cause,
- }),
- ),
- );
+ const catchUpStream = orchestrationEngine
+ .readEvents(input.afterSequence, Number.MAX_SAFE_INTEGER)
+ .pipe(
+ Stream.filter(isThisThreadDetailEvent),
+ Stream.map((event) => ({ kind: "event" as const, event })),
+ Stream.mapError(
+ (cause) =>
+ new OrchestrationGetSnapshotError({
+ message: `Failed to replay thread ${input.threadId} events`,
+ cause,
+ }),
+ ),
+ );
return Stream.concat(catchUpStream, liveStream);
}
- const [threadDetail, snapshotSequence] = yield* Effect.all([
- projectionSnapshotQuery.getThreadDetailById(input.threadId).pipe(
+ const threadSnapshot = yield* projectionSnapshotQuery
+ .getThreadDetailSnapshot(input.threadId)
+ .pipe(
Effect.mapError(
(cause) =>
new OrchestrationGetSnapshotError({
@@ -1158,20 +1166,9 @@
cause,
}),
),
- ),
- projectionSnapshotQuery.getSnapshotSequence().pipe(
- Effect.map(({ snapshotSequence }) => snapshotSequence),
- Effect.mapError(
- (cause) =>
- new OrchestrationGetSnapshotError({
- message: "Failed to load orchestration snapshot sequence",
- cause,
- }),
- ),
- ),
- ]);
+ );
- if (Option.isNone(threadDetail)) {
+ if (Option.isNone(threadSnapshot)) {
return yield* new OrchestrationGetSnapshotError({
message: `Thread ${input.threadId} was not found`,
cause: input.threadId,
@@ -1181,10 +1178,7 @@
return Stream.concat(
Stream.make({
kind: "snapshot" as const,
- snapshot: {
- snapshotSequence,
- thread: threadDetail.value,
- },
+ snapshot: threadSnapshot.value,
}),
liveStream,
);You can send follow-ups to the cloud agent here.
Contributor
ApprovabilityVerdict: Needs human review 1 blocking correctness issue found. This PR introduces a new synchronization strategy (HTTP snapshot loading before WebSocket sync) with significant runtime behavior changes across server and client layers. Two unresolved medium-severity review comments identify potential bugs in the new sync behavior that should be addressed. You can customize Macroscope's approvability policy. Learn more. |
- ws.ts (afterSequence resume): attach the live PubSub subscription into a scope-bound buffer BEFORE draining the catch-up replay, so events published during the replay window are not lost; read the full range after the cursor (readEvents limit) so the per-thread filter can't be starved by a global cap. - ProjectionSnapshotQuery: add getThreadDetailSnapshot that reads thread detail + snapshot sequence in one transaction, so the sequence is consistent with the returned state; use it in the HTTP handler and the WS snapshot path. - OrchestrationEngine.readEvents: accept an optional limit. - client loader: bound the HTTP snapshot timeout (cached data renders meanwhile) and treat a 404 as an expected defer-to-socket case rather than a warning. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
The loader layer resolved ManagedRelayDpopSigner eagerly at build, hard-requiring it even though only relay/DPoP connections need it. Resolve it via Effect.serviceOption and pass it through to buildAuthHeaders, so the layer only requires HttpClient; bearer/primary connections no longer depend on a signer. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
Effect service conventions require catchTags (object form) over catchTag even for a single tag. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
…napshot Part 1 (thread warm-cache): persist the snapshot sequence alongside the cached thread (StoredThreadSnapshot v2) so a re-opened thread resumes from the cached sequence via afterSequence — receiving only events since then — instead of re-downloading its full body. Cold cache still loads the full snapshot over HTTP. Part 2 (shell): the shell subscription now resumes the same way. Added an afterSequence input to subscribeShell (server replays shell events after the cursor, buffering live events before the catch-up replay, as in the thread path) and a GET /api/orchestration/shell endpoint so a cold-cache full shell rides HTTP instead of the socket. The client passes the cached/loaded shell sequence. Shared the environment auth-header builder across the thread and shell HTTP loaders. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
github-actions
Bot
added
size:XL
Contributor
There was a problem hiding this comment.
Cursor Bugbot has reviewed your changes using high effort and found 1 potential issue.
Bugbot Autofix is ON. A cloud agent has been kicked off to fix the reported issue. You can view the agent here.
Reviewed by Cursor Bugbot for commit 9bc191b. Configure here.
| cause, | ||
| }), | ||
| ), | ||
| ); |
Contributor
There was a problem hiding this comment.
Shell catch-up uses current DB
Medium Severity
When subscribeShell resumes with afterSequence, catch-up replays persisted events through toShellStreamEvent, which loads project/thread shells from the current projection (getProjectShellById / getThreadShellById). Those reads reflect today’s state, not the state at each replayed sequence, so the client can apply the wrong shell data (or drop upserts for deleted rows) while still advancing snapshotSequence.
Additional Locations (1)
Reviewed by Cursor Bugbot for commit 9bc191b. Configure here.
Contributor
There was a problem hiding this comment.
Bugbot Autofix determined this is a false positive.
The shell stream replicates current state (whole-row upserts deduped by monotonic sequence), and because event append and projection update commit in one transaction, replayed events always carry data at a version >= their sequence while every deletion/archive emits its own payload-derived removal event later in the same replay, so the client provably converges to the server's current shell with no dropped changes — the same read-current-projection semantics the pre-existing live path already uses.
You can send follow-ups to the cloud agent here.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
1 participant
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Summary
Testing
vp checkvp run typecheckvp run lintvp test/vp run testfor the updated client-runtime state testsvp run lint:mobile)Note
Medium Risk
Touches live orchestration sync (HTTP + WebSocket resume, event replay ordering) and projection read consistency; mistakes could cause missed or duplicated thread/shell updates, though fallbacks and sequence dedup mitigate some failure modes.
Overview
Orchestration sync now prefers HTTP for initial shell and thread snapshots, then WebSocket catch-up from a known sequence instead of embedding full snapshots on the socket.
Clients use new
ShellSnapshotLoader/ThreadSnapshotLoaderservices (Bearer/DPoP auth, bounded timeouts, socket fallback on failure). Warm caches (v2 thread cache storesOrchestrationThreadDetailSnapshot+ sequence) subscribe withafterSequenceand skip HTTP; cold paths fetchGET /api/orchestration/shellandGET /api/orchestration/threads/:threadId(404 →thread_not_found).Server:
getThreadDetailSnapshotreads thread detail and global snapshot sequence in one transaction;subscribeShell/subscribeThreadaccept optionalafterSequence, buffer live events before replay, and usereadEvents(..., limit)without the default page cap so per-thread/shell filters do not drop events. Web and mobile connection runtimes merge the snapshot loader layers.Contracts add
OrchestrationSubscribeShellInput,EnvironmentResourceNotFoundError, and wire HTTP + RPC types; tests cover HTTP seeding, warm-cache resume, and deduped replay.Reviewed by Cursor Bugbot for commit 9bc191b. Bugbot is set up for automated code reviews on this repo. Configure here.
Note
Load thread and shell snapshots over HTTP before opening WebSocket subscriptions
GET /api/orchestration/shellandGET /api/orchestration/threads/:threadIdon the server, backed byProjectionSnapshotQuery.getThreadDetailSnapshotwhich reads thread detail and snapshot sequence atomically.makeEnvironmentShellStateandmakeEnvironmentThreadStatenow attempt to load a snapshot from cache or HTTP before opening the WebSocket, then passafterSequencetosubscribeShell/subscribeThreadto resume from that point instead of receiving a full snapshot over the socket.subscribeShellandsubscribeThreadWebSocket handlers now supportafterSequence: when provided, they replay persisted events after that sequence and then stream live events, skipping the initial snapshot frame.OrchestrationThreadDetailSnapshot(thread +snapshotSequence) instead of a bare thread object.EnvironmentResourceNotFoundError(HTTP 404) is introduced for missing threads and is handled gracefully client-side (404 on thread snapshot →Option.none, treated as a cold-cache start).Macroscope summarized 9bc191b.