Skip to content

Commit f9b5fec

Browse files
juliusmarmingeclaude
juliusmarminge
and
claude
committed
perf(client): reopening a finished command row no longer refetches its output
Co-Authored-By: Claude Opus 5.5 <[email protected]>
1 parent 1fed70c commit f9b5fec

3 files changed

Lines changed: 122 additions & 18 deletions

File tree

‎packages/client-runtime/src/state/orchestration.ts‎

Lines changed: 31 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -4,6 +4,7 @@ import {
44
type ThreadId,
55
type OrchestrationV2ThreadLinkMetadata,
66
type OrchestrationGetTurnDiffResult,
7+
type OrchestrationV2GetTurnItemInput,
78
} from "@t3tools/contracts";
89
import * as Effect from "effect/Effect";
910
import * as Option from "effect/Option";
@@ -12,7 +13,11 @@ import * as SubscriptionRef from "effect/SubscriptionRef";
1213
import type { HttpClient } from "effect/http";
1314
import { Atom } from "effect/reactivity";
1415

15-
import { applyCommandOutputFrame, EMPTY_COMMAND_OUTPUT } from "./commandOutput.ts";
16+
import {
17+
applyCommandOutputFrame,
18+
EMPTY_COMMAND_OUTPUT,
19+
type CommandOutputView,
20+
} from "./commandOutput.ts";
1621

1722
import {
1823
createEnvironmentQueryAtomFamily,
@@ -49,6 +54,19 @@ export function createOrchestrationEnvironmentAtoms<R, E>(
4954
? {}
5055
: { onAnswer: options.onCheckpointDiffAnswer };
5156
const shellState = options.shellState;
57+
const liveTurnItem = createEnvironmentRpcQueryAtomFamily(runtime, {
58+
label: "environment-data:orchestration:turn-item",
59+
tag: ORCHESTRATION_V2_WS_METHODS.getTurnItem,
60+
staleTimeMs: 60_000,
61+
idleTtlMs: 60_000,
62+
});
63+
// A settled item's revision is its updatedAt, so the answer for it never goes stale.
64+
const settledTurnItem = createEnvironmentRpcQueryAtomFamily(runtime, {
65+
label: "environment-data:orchestration:turn-item",
66+
tag: ORCHESTRATION_V2_WS_METHODS.getTurnItem,
67+
staleTimeMs: Number.POSITIVE_INFINITY,
68+
idleTtlMs: 5 * 60_000,
69+
});
5270
const refreshThreadLookup = ({
5371
environmentId,
5472
input,
@@ -139,12 +157,15 @@ export function createOrchestrationEnvironmentAtoms<R, E>(
139157
tag: ORCHESTRATION_V2_WS_METHODS.subscribeThread,
140158
idleTtlMs: 0,
141159
}),
142-
// One expanded command row's output. Mounted only while the row is open,
143-
// and released as soon as it closes, so hidden rows receive nothing.
160+
// One expanded command row's output. A running command's subscription ends
161+
// as soon as its row closes, so hidden rows receive nothing; a finished
162+
// command's final output stays cached so reopening it does not refetch.
144163
commandOutput: createEnvironmentRpcSubscriptionAtomFamily(runtime, {
145164
label: "environment-data:orchestration-v2:command-output",
146165
tag: ORCHESTRATION_V2_WS_METHODS.subscribeCommandOutput,
147166
idleTtlMs: 0,
167+
completeWhen: (view: CommandOutputView) => !view.running,
168+
completedTtlMs: 5 * 60_000,
148169
// `scan` emits its seed first; skipping it keeps the row "loading" until the first frame.
149170
transform: (frames) =>
150171
frames.pipe(
@@ -185,12 +206,13 @@ export function createOrchestrationEnvironmentAtoms<R, E>(
185206
label: "environment-data:orchestration:force-stop-thread",
186207
tag: ORCHESTRATION_V2_WS_METHODS.forceStopThread,
187208
}),
188-
turnItem: createEnvironmentRpcQueryAtomFamily(runtime, {
189-
label: "environment-data:orchestration:turn-item",
190-
tag: ORCHESTRATION_V2_WS_METHODS.getTurnItem,
191-
staleTimeMs: 60_000,
192-
idleTtlMs: 60_000,
193-
}),
209+
turnItem: (target: {
210+
readonly environmentId: EnvironmentId;
211+
readonly input: OrchestrationV2GetTurnItemInput;
212+
}) =>
213+
target.input.revision === undefined || target.input.revision === "live"
214+
? liveTurnItem(target)
215+
: settledTurnItem(target),
194216
fullThreadDiff: createEnvironmentRpcQueryAtomFamily(runtime, {
195217
label: "environment-data:orchestration:full-thread-diff",
196218
tag: ORCHESTRATION_V2_WS_METHODS.getFullThreadDiff,

‎packages/client-runtime/src/state/runtime.test.ts‎

Lines changed: 57 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -392,6 +392,63 @@ describe("finite environment subscription lifecycle", () => {
392392
}),
393393
),
394394
);
395+
it.effect.each([
396+
{ completedTtlMs: undefined, subscriptions: 2 },
397+
{ completedTtlMs: 60_000, subscriptions: 1 },
398+
])(
399+
"reopening a completed subscription subscribes $subscriptions time(s) with completedTtlMs $completedTtlMs",
400+
({ completedTtlMs, subscriptions }) =>
401+
Effect.gen(function* () {
402+
let subscribed = 0;
403+
const environmentRegistry = EnvironmentRegistry.EnvironmentRegistry.of({
404+
followStream: (_environmentId: EnvironmentId, source: Stream.Stream<unknown>) => source,
405+
} as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]);
406+
const runtime = Atom.runtime(
407+
Layer.succeed(EnvironmentRegistry.EnvironmentRegistry, environmentRegistry),
408+
);
409+
const family = createEnvironmentSubscriptionAtomFamily(runtime, {
410+
label: "test.completed-subscription",
411+
subscribe: () =>
412+
Stream.suspend(() => {
413+
subscribed += 1;
414+
return Stream.make({ complete: false }, { complete: true });
415+
}),
416+
completeWhen: (value) => value.complete,
417+
idleTtlMs: 0,
418+
...(completedTtlMs === undefined ? {} : { completedTtlMs }),
419+
});
420+
const atom = family({ environmentId: QUERY_ENVIRONMENT.environmentId, input: undefined });
421+
// Flushed after each close: idle removals have run by the time a user reopens a row.
422+
const scheduled: Array<() => void> = [];
423+
const registry = AtomRegistry.make({
424+
scheduleTask: (task) => {
425+
scheduled.push(task);
426+
return () => undefined;
427+
},
428+
});
429+
const open = Effect.gen(function* () {
430+
const settled = yield* Deferred.make<boolean>();
431+
const unsubscribe = registry.subscribe(
432+
atom,
433+
(result) => {
434+
if (AsyncResult.isSuccess(result) && !result.waiting) {
435+
Deferred.doneUnsafe(settled, Exit.succeed(result.value.complete));
436+
}
437+
},
438+
{ immediate: true },
439+
);
440+
const complete = yield* Deferred.await(settled);
441+
unsubscribe();
442+
for (const task of scheduled.splice(0)) task();
443+
return complete;
444+
});
445+
446+
expect(yield* open).toBe(true);
447+
expect(yield* open).toBe(true);
448+
expect(subscribed).toBe(subscriptions);
449+
registry.dispose();
450+
}),
451+
);
395452
});
396453

397454
describe("environment query lifecycle", () => {

‎packages/client-runtime/src/state/runtime.ts‎

Lines changed: 34 additions & 9 deletions
Original file line numberDiff line numberDiff line change
@@ -122,6 +122,8 @@ interface EnvironmentSubscriptionAtomOptions<Input, A, E, R> {
122122
readonly subscribe: (input: Input) => Stream.Stream<A, E, R>;
123123
readonly completeWhen?: (value: A) => boolean;
124124
readonly idleTtlMs?: number;
125+
/** Keeps a completed value this long after its last reader leaves, so reopening skips a resubscribe. */
126+
readonly completedTtlMs?: number;
125127
}
126128

127129
export type SettledAsyncResult<A, E> = AsyncResult.Success<A, E> | AsyncResult.Failure<A, E>;
@@ -826,23 +828,44 @@ export function createEnvironmentSubscriptionAtomFamily<R, ER, Input, A, E>(
826828
EnvironmentSupervisor.EnvironmentSupervisor | R
827829
>,
828830
) {
829-
const family = Atom.family((key: string) => {
831+
const idleTtlMs = options.idleTtlMs ?? 5 * 60_000;
832+
const label = (key: string) =>
833+
options.sensitiveInput
834+
? `${options.label}:${parseEnvironmentRpcKey<Input>(key).environmentId}`
835+
: `${options.label}:${key}`;
836+
const live = Atom.family((key: string) => {
830837
const target = parseEnvironmentRpcKey<Input>(key);
831838
return runtime
832839
.atom(
833840
Stream.suspend(() =>
834841
followStreamInEnvironment(target.environmentId, options.subscribe(target.input)),
835842
).pipe(options.completeWhen ? Stream.takeUntil(options.completeWhen) : (stream) => stream),
836843
)
837-
.pipe(
838-
Atom.setIdleTTL(options.idleTtlMs ?? 5 * 60_000),
839-
Atom.withLabel(
840-
options.sensitiveInput
841-
? `${options.label}:${target.environmentId}`
842-
: `${options.label}:${key}`,
843-
),
844-
);
844+
.pipe(Atom.setIdleTTL(idleTtlMs), Atom.withLabel(label(key)));
845845
});
846+
const { completeWhen, completedTtlMs } = options;
847+
if (completeWhen === undefined || completedTtlMs === undefined) {
848+
return (target: { readonly environmentId: EnvironmentIdType; readonly input: Input }) =>
849+
live(environmentRpcKey(target));
850+
}
851+
const completed = Atom.family((key: string) =>
852+
Atom.make(Option.none<A>()).pipe(
853+
Atom.setIdleTTL(completedTtlMs),
854+
Atom.withLabel(`${label(key)}:completed`),
855+
),
856+
);
857+
// Reading `completed` first drops the live subscription once its value is final.
858+
const family = Atom.family((key: string) =>
859+
Atom.make((get) => {
860+
const done = get(completed(key));
861+
if (Option.isSome(done)) return AsyncResult.success(done.value);
862+
const result = get(live(key));
863+
if (AsyncResult.isSuccess(result) && !result.waiting && completeWhen(result.value)) {
864+
get.set(completed(key), Option.some(result.value));
865+
}
866+
return result;
867+
}).pipe(Atom.setIdleTTL(idleTtlMs), Atom.withLabel(label(key))),
868+
);
846869
return (target: { readonly environmentId: EnvironmentIdType; readonly input: Input }) =>
847870
family(environmentRpcKey(target));
848871
}
@@ -929,6 +952,7 @@ export function createEnvironmentRpcSubscriptionAtomFamily<
929952
readonly tag: TTag;
930953
readonly idleTtlMs?: number;
931954
readonly completeWhen?: (value: B) => boolean;
955+
readonly completedTtlMs?: number;
932956
readonly transform?: (
933957
stream: Stream.Stream<
934958
EnvironmentRpcStreamValue<TTag>,
@@ -945,6 +969,7 @@ export function createEnvironmentRpcSubscriptionAtomFamily<
945969
return createEnvironmentSubscriptionAtomFamily(runtime, {
946970
label: options.label,
947971
...(options.completeWhen === undefined ? {} : { completeWhen: options.completeWhen }),
972+
...(options.completedTtlMs === undefined ? {} : { completedTtlMs: options.completedTtlMs }),
948973
...(options.idleTtlMs === undefined ? {} : { idleTtlMs: options.idleTtlMs }),
949974
subscribe: (input: EnvironmentRpcInput<TTag>) => {
950975
const stream = subscribe(options.tag, input);

0 commit comments

Comments
 (0)