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

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
49 changes: 46 additions & 3 deletions apps/web/src/state/terminalSessions.test.ts
Original file line number Diff line number Diff line change
@@ -1,11 +1,23 @@
import { describe, expect, it, vi } from "vite-plus/test";
import { EnvironmentId, ThreadId, type TerminalSummary } from "@t3tools/contracts";
import { selectRunningSubprocessTerminalIds } from "@t3tools/client-runtime/state/terminal";
import { AtomRegistry } from "effect/reactivity";
import * as Cause from "effect/Cause";
import * as Option from "effect/Option";
import { AsyncResult, AtomRegistry } from "effect/reactivity";

import { attachedTerminalSessionAtom, selectKnownTerminalSessions } from "./terminalSessions";
import {
attachedTerminalSessionAtom,
knownTerminalSessionsAtom,
selectKnownTerminalSessions,
} from "./terminalSessions";

const attachCalls = vi.hoisted(() => [] as unknown[]);
const metadataAtom = await vi.hoisted(async () => {
const { AsyncResult, Atom } = await import("effect/reactivity");
return Atom.make<AsyncResult.AsyncResult<ReadonlyArray<TerminalSummary>>>(
AsyncResult.initial(true),
);
});
vi.mock("./terminal", async () => {
const { AsyncResult, Atom } = await import("effect/reactivity");
return {
Expand All @@ -14,7 +26,7 @@ vi.mock("./terminal", async () => {
attachCalls.push(request);
return Atom.make(AsyncResult.initial(true));
},
metadata: () => Atom.make(AsyncResult.initial(true)),
metadata: () => metadataAtom,
},
};
});
Expand Down Expand Up @@ -100,6 +112,37 @@ describe("selectKnownTerminalSessions", () => {
});
});

describe("knownTerminalSessionsAtom", () => {
const metadataResult = (...terminals: ReadonlyArray<TerminalSummary>) =>
AsyncResult.success<ReadonlyArray<TerminalSummary>>(terminals);

it("notifies only when the thread's own terminals change", () => {
const registry = AtomRegistry.make();
const sessions = knownTerminalSessionsAtom(environmentA, threadA);
expect(registry.get(sessions)).toBeNull();

const first = summary(threadA, "terminal-1");
const other = summary(threadB, "terminal-1");
registry.set(metadataAtom, metadataResult(first));
const notified: unknown[] = [];
const unsubscribe = registry.subscribe(sessions, (value) => notified.push(value));
const before = registry.get(sessions);

registry.set(metadataAtom, metadataResult(first, other));
registry.set(metadataAtom, metadataResult(first, { ...other, label: "Renamed" }));
expect(notified).toEqual([]);
expect(registry.get(sessions)).toBe(before);

registry.set(metadataAtom, metadataResult({ ...first, label: "Renamed" }, other));
expect(notified).toHaveLength(1);

const previousSuccess = Option.some(metadataResult(first));
registry.set(metadataAtom, AsyncResult.failure(Cause.die("unavailable"), { previousSuccess }));
expect(registry.get(sessions)).toBeNull();
unsubscribe();
});
});

describe("attachedTerminalSessionAtom", () => {
it("attaches with the launch input it was keyed by", () => {
attachCalls.length = 0;
Expand Down
39 changes: 29 additions & 10 deletions apps/web/src/state/terminalSessions.ts
Original file line number Diff line number Diff line change
@@ -1,3 +1,4 @@
import { useAtomValue } from "@effect/atom-react";
import {
combineTerminalSessionState,
EMPTY_TERMINAL_BUFFER_STATE,
Expand All @@ -19,7 +20,7 @@ import * as Option from "effect/Option";
import * as Schema from "effect/Schema";
import { AsyncResult, Atom } from "effect/reactivity";

import { formatEnvironmentQueryError, useEnvironmentQuery } from "./query";
import { formatEnvironmentQueryError } from "./query";
import { terminalEnvironment } from "./terminal";
import { environmentSession, sessionHasScope, useEnvironmentScope } from "./session";

Expand Down Expand Up @@ -194,22 +195,40 @@ export function attachedTerminalSessionAtom(
return attachedTerminalSessionFamily(JSON.stringify([environmentId, terminal]));
}

const NO_KNOWN_TERMINAL_SESSIONS_ATOM = Atom.make<ReadonlyArray<KnownTerminalSession> | null>(null);

const decodeKnownSessionsKey = Schema.decodeUnknownOption(
Schema.fromJsonString(Schema.Tuple([EnvironmentId, Schema.NullOr(ThreadId)])),
);

// Derived per thread, so a terminal update elsewhere in the environment notifies no one here.
const knownTerminalSessionsFamily = Atom.family((key: string) => {
const [environmentId, threadId] = Option.getOrThrow(decodeKnownSessionsKey(key));
return Atom.make((get) => {
const metadata = get(terminalEnvironment.metadata({ environmentId, input: null }));
return metadata._tag === "Success"
? selectKnownTerminalSessions(metadata.value, environmentId, threadId)
: null;
}).pipe(Atom.withLabel(`web-known-terminal-sessions:${key}`));
});

export function knownTerminalSessionsAtom(
environmentId: EnvironmentId,
threadId: ThreadId | null,
): Atom.Atom<ReadonlyArray<KnownTerminalSession> | null> {
return knownTerminalSessionsFamily(JSON.stringify([environmentId, threadId]));
}

export function useKnownTerminalSessions(input: {
readonly environmentId: EnvironmentId | null;
readonly threadId: ThreadId | null;
}): ReadonlyArray<KnownTerminalSession> | null {
const canRead = useEnvironmentScope(input.environmentId, AuthTerminalReadScope);
const metadata = useEnvironmentQuery(
return useAtomValue(
input.environmentId === null || !canRead
? null
: terminalEnvironment.metadata({
environmentId: input.environmentId,
input: null,
}),
? NO_KNOWN_TERMINAL_SESSIONS_ATOM
: knownTerminalSessionsAtom(input.environmentId, input.threadId),
);
return metadata.data === null || metadata.error !== null
? null
: selectKnownTerminalSessions(metadata.data, input.environmentId, input.threadId);
}

export function useThreadRunningTerminalIds(input: {
Expand Down
33 changes: 33 additions & 0 deletions apps/web/test/appScenarios.renders.tsx
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@ import { getRouter, type AppRouter } from "~/router";
import {
ENVIRONMENT_ID,
assistantItem,
emitTerminalMetadata,
emitThreadEvent,
installFakeEnvironment,
isThreadSubscribed,
Expand Down Expand Up @@ -50,6 +51,7 @@ vi.mock(import("~/rpc/requestLatencyState"), async (original) => ({
const HISTORY_TURNS = 20;
const STREAM_DELTAS = 30;
const TYPED_TEXT = "fix the flaky tests!";
const TERMINAL_UPDATES = 20;
// The stream scenario leaves a turn on its thread, so it streams into one no other scenario opens.
const THREADS = ["thread-warm", "thread-a", "thread-b", "thread-live"].map((name) =>
threadWithHistory(name, HISTORY_TURNS),
Expand Down Expand Up @@ -211,6 +213,37 @@ it("app-stream-assistant-turn", async () => {
await report("app-stream-assistant-turn", [counted], { commits: false });
});

it("app-terminal-updates-in-other-thread", async () => {
using _clock = frozenClock();
await using _app = (await openThread(A)).screen;
const counted = await measure(async () => {
for (let update = 0; update < TERMINAL_UPDATES; update++) {
await act(() =>
emitTerminalMetadata({
type: "upsert",
terminal: {
threadId: B,
terminalId: "default",
cwd: "/repo",
worktreePath: null,
status: "running",
pid: 42,
exitCode: null,
exitSignal: null,
hasRunningSubprocess: true,
label: `command ${update}`,
updatedAt: `2026-03-18T09:01:${String(update).padStart(2, "0")}.000Z`,
},
}),
);
}
});
// Only thread B's sidebar row shows these updates.
expect(counted.components.ChatView).toBeUndefined();
expect(counted.components.SidebarThreadRow).toBe(TERMINAL_UPDATES);
await report("app-terminal-updates-in-other-thread", [counted]);
});

it("app-type-in-composer", async () => {
using _clock = frozenClock();
await using _app = (await openThread(A)).screen;
Expand Down
14 changes: 14 additions & 0 deletions apps/web/test/fakeEnvironment.ts
Original file line number Diff line number Diff line change
Expand Up @@ -25,6 +25,7 @@ import {
type OrchestrationV2ThreadShell,
type OrchestrationV2ThreadStreamItem,
type OrchestrationV2TurnItem,
type TerminalMetadataStreamEvent,
} from "@t3tools/contracts";
import type {
ConnectionAttemptError,
Expand Down Expand Up @@ -307,6 +308,7 @@ const json = (body: unknown, status = 200) =>

// Keyed by thread id: a live thread subscription's queue, for pushing events to it.
const threadSubscribers = new Map<string, Queue.Queue<OrchestrationV2ThreadStreamItem>>();
let terminalMetadataSubscriber: Queue.Queue<TerminalMetadataStreamEvent> | null = null;
let threads: ReadonlyArray<FakeThread> = [];
let sequence = 1;

Expand All @@ -324,6 +326,13 @@ const subscribe = <A>(first: ReadonlyArray<A>, register?: (queue: Queue.Queue<A>
);

const handlers: Record<string, (input: never) => unknown> = {
[WS_METHODS.subscribeTerminalMetadata]: () =>
subscribe<TerminalMetadataStreamEvent>([{ type: "snapshot", terminals: [] }], (queue) => {
terminalMetadataSubscriber = queue;
return () => {
if (terminalMetadataSubscriber === queue) terminalMetadataSubscriber = null;
};
}),
[ORCHESTRATION_V2_WS_METHODS.subscribeShell]: () =>
subscribe<OrchestrationV2ShellStreamItem>([
{
Expand Down Expand Up @@ -437,6 +446,11 @@ export function installFakeEnvironment(fixtures: ReadonlyArray<FakeThread>): Dis

type ThreadEvent = Extract<OrchestrationV2ThreadStreamItem, { readonly kind: "event" }>["event"];

export async function emitTerminalMetadata(event: TerminalMetadataStreamEvent): Promise<void> {
if (terminalMetadataSubscriber === null) throw new Error("No terminal metadata subscription");
await Effect.runPromise(Queue.offer(terminalMetadataSubscriber, event));
}

/** Whether the app holds a live subscription to the thread, so events reach it. */
export function isThreadSubscribed(threadId: ThreadId): boolean {
return threadSubscribers.has(threadId);
Expand Down
Loading