perf(client-runtime): batch terminal input writes

apps/mobile/src/features/terminal/ThreadTerminalRouteScreen.tsx

@@ -161,6 +161,7 @@ type ThreadTerminalRouteScreenProps = StaticScreenProps<{
 export function ThreadTerminalRouteScreen(props: ThreadTerminalRouteScreenProps) {
   const navigation = useNavigation();
   const writeTerminal = useAtomCommand(terminalEnvironment.write, "terminal write");
+  const writeTerminalInput = useAtomCommand(terminalEnvironment.input, "terminal input");
   const resizeTerminal = useAtomCommand(terminalEnvironment.resize, "terminal resize");
   const clearTerminal = useAtomCommand(terminalEnvironment.clear, "terminal clear");
   const closeTerminal = useAtomCommand(terminalEnvironment.close, "terminal close");
@@ -699,7 +700,7 @@ export function ThreadTerminalRouteScreen(props: ThreadTerminalRouteScreenProps)
         return;
       }
 
-      void writeTerminal({
+      void writeTerminalInput({
         environmentId: selectedThread.environmentId,
         input: {
           threadId: selectedThread.id,
@@ -708,7 +709,7 @@ export function ThreadTerminalRouteScreen(props: ThreadTerminalRouteScreenProps)
         },
       });
     },
-    [isRunning, selectedThread, terminalId, writeTerminal],
+    [isRunning, selectedThread, terminalId, writeTerminalInput],
   );
 
   const handleInput = useCallback(

apps/web/src/components/ThreadTerminalDrawer.tsx

@@ -322,7 +322,7 @@ export function TerminalViewport({
   const openPreview = useAtomCommand(previewEnvironment.open, {
     reportFailure: false,
   });
-  const runTerminalWrite = useAtomCommand(terminalEnvironment.write, {
+  const runTerminalWrite = useAtomCommand(terminalEnvironment.input, {
     reportFailure: false,
   });
   const runTerminalResize = useAtomCommand(terminalEnvironment.resize, {

packages/client-runtime/src/rpc/client.ts

@@ -145,6 +145,32 @@ export const request = Effect.fn("EnvironmentRpc.request")(function* <
   return yield* method(input).pipe(Effect.ensuring(completeObservation));
 });
 
+/**
+ * Sends a unary RPC without waiting for a response frame. This is reserved for
+ * ordered, ephemeral traffic where replay would be unsafe and the caller has a
+ * separate authoritative stream for observing the resulting state.
+ */
+export const requestDiscard = Effect.fn("EnvironmentRpc.requestDiscard")(function* <
+  TTag extends EnvironmentUnaryRpcTag,
+>(tag: TTag, input: EnvironmentRpcInput<TTag>) {
+  const supervisor = yield* EnvironmentSupervisor;
+  yield* Effect.annotateCurrentSpan({
+    "environment.id": supervisor.target.environmentId,
+    "rpc.method": tag,
+  });
+  const session = yield* currentSession();
+  const observer = yield* EnvironmentRpcRequestObserver;
+  const method = session.client[tag] as (
+    input: EnvironmentRpcInput<TTag>,
+    options: { readonly discard: true },
+  ) => Effect.Effect<void, RpcClientError.RpcClientError>;
+  const completeObservation = yield* observer.observe({
+    environmentId: supervisor.target.environmentId,
+    method: tag,
+  });
+  return yield* method(input, { discard: true }).pipe(Effect.ensuring(completeObservation));
+});
+
 export function runStream<TTag extends EnvironmentStreamCommandRpcTag>(
   tag: TTag,
   input: EnvironmentRpcInput<TTag>,

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

@@ -1,15 +1,20 @@
 import { type TerminalSummary, WS_METHODS } from "@t3tools/contracts";
+import * as Effect from "effect/Effect";
 import * as Stream from "effect/Stream";
-import { Atom } from "effect/unstable/reactivity";
+import { AsyncResult, Atom } from "effect/unstable/reactivity";
 
 import {
   createAtomCommandScheduler,
+  createEnvironmentCommand,
   createEnvironmentRpcCommand,
   createEnvironmentRpcSubscriptionAtomFamily,
   createEnvironmentSubscriptionAtomFamily,
+  runInEnvironment,
+  settleAsyncResult,
 } from "./runtime.ts";
 import type { EnvironmentRegistry } from "../connection/registry.ts";
-import { subscribe, type EnvironmentRpcInput } from "../rpc/client.ts";
+import { requestDiscard, subscribe, type EnvironmentRpcInput } from "../rpc/client.ts";
+import { createTerminalInputCommand } from "./terminalInput.ts";
 import {
   applyTerminalAttachStreamEvent,
   applyTerminalMetadataStreamEvent,
@@ -36,6 +41,28 @@ export function createTerminalEnvironmentAtoms<R, E>(
     readonly input: { readonly threadId: string; readonly terminalId?: string | undefined };
   }) => JSON.stringify([environmentId, input.threadId, input.terminalId ?? null]);
   const lifecycleConcurrency = { mode: "serial" as const, key: terminalThreadKey };
+  const sendTerminalInputFallback = createEnvironmentCommand(runtime, {
+    label: "environment-data:terminal:input:send",
+    execute: (input: EnvironmentRpcInput<typeof WS_METHODS.terminalWrite>) =>
+      requestDiscard(WS_METHODS.terminalWrite, input),
+  });
+  const sendTerminalInput: typeof sendTerminalInputFallback = {
+    label: sendTerminalInputFallback.label,
+    run: (registry, target) => {
+      const runtimeContext = registry.get(runtime);
+      if (!AsyncResult.isSuccess(runtimeContext)) {
+        return sendTerminalInputFallback.run(registry, target);
+      }
+      return settleAsyncResult(() =>
+        Effect.runPromiseExitWith(runtimeContext.value)(
+          runInEnvironment(
+            target.environmentId,
+            requestDiscard(WS_METHODS.terminalWrite, target.input),
+          ),
+        ),
+      );
+    },
+  };
   return {
     attach: createEnvironmentSubscriptionAtomFamily(runtime, {
       label: "environment-data:terminal:attach",
@@ -65,6 +92,7 @@ export function createTerminalEnvironmentAtoms<R, E>(
       label: "environment-data:terminal:write",
       tag: WS_METHODS.terminalWrite,
     }),
+    input: createTerminalInputCommand(sendTerminalInput),
     resize: createEnvironmentRpcCommand(runtime, {
       label: "environment-data:terminal:resize",
       tag: WS_METHODS.terminalResize,
@@ -93,3 +121,4 @@ export function createTerminalEnvironmentAtoms<R, E>(
 }
 
 export * from "./terminalSession.ts";
+export * from "./terminalInput.ts";

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

@@ -0,0 +1,190 @@
+import { describe, expect, it } from "vite-plus/test";
+
+import { EnvironmentId, ThreadId } from "@t3tools/contracts";
+import * as Cause from "effect/Cause";
+import * as AsyncResult from "effect/unstable/reactivity/AsyncResult";
+import { AtomRegistry } from "effect/unstable/reactivity";
+
+import {
+  createTerminalInputCommand,
+  TerminalInputBackpressureError,
+  type TerminalInputTarget,
+} from "./terminalInput.ts";
+import type { AtomCommand } from "./runtime.ts";
+
+const target = (data: string, terminalId = "term-1"): TerminalInputTarget => ({
+  environmentId: EnvironmentId.make("environment-1"),
+  input: {
+    threadId: ThreadId.make("thread-1"),
+    terminalId,
+    data,
+  },
+});
+
+describe("terminal input command", () => {
+  it("coalesces same-task input and preserves its order", async () => {
+    const sent: string[] = [];
+    const send: AtomCommand<TerminalInputTarget, void, never> = {
+      label: "send",
+      run: async (_registry, input) => {
+        sent.push(input.input.data);
+        return AsyncResult.success(undefined);
+      },
+    };
+    const input = createTerminalInputCommand(send);
+    const registry = AtomRegistry.make();
+
+    const results = await Promise.all([
+      input.run(registry, target("a")),
+      input.run(registry, target("b")),
+      input.run(registry, target("c")),
+    ]);
+
+    expect(sent).toEqual(["abc"]);
+    expect(results.every(AsyncResult.isSuccess)).toBe(true);
+    registry.dispose();
+  });
+
+  it("keeps independent terminal lanes separate", async () => {
+    const sent: Array<{ terminalId: string; data: string }> = [];
+    const send: AtomCommand<TerminalInputTarget, void, never> = {
+      label: "send",
+      run: async (_registry, input) => {
+        sent.push({ terminalId: input.input.terminalId, data: input.input.data });
+        return AsyncResult.success(undefined);
+      },
+    };
+    const input = createTerminalInputCommand(send);
+    const registry = AtomRegistry.make();
+
+    await Promise.all([
+      input.run(registry, target("a", "term-1")),
+      input.run(registry, target("b", "term-2")),
+    ]);
+
+    expect(sent).toEqual([
+      { terminalId: "term-1", data: "a" },
+      { terminalId: "term-2", data: "b" },
+    ]);
+    registry.dispose();
+  });
+
+  it("splits queued input into ordered batches at the configured size", async () => {
+    const sent: string[] = [];
+    const send: AtomCommand<TerminalInputTarget, void, never> = {
+      label: "send",
+      run: async (_registry, input) => {
+        sent.push(input.input.data);
+        return AsyncResult.success(undefined);
+      },
+    };
+    const input = createTerminalInputCommand(send, { maxBatchChars: 3 });
+    const registry = AtomRegistry.make();
+
+    const results = await Promise.all([
+      input.run(registry, target("ab")),
+      input.run(registry, target("cd")),
+      input.run(registry, target("e")),
+    ]);
+
+    expect(sent).toEqual(["ab", "cde"]);
+    expect(results.every(AsyncResult.isSuccess)).toBe(true);
+    registry.dispose();
+  });
+
+  it("splits a single oversized input into legal ordered writes", async () => {
+    const sent: string[] = [];
+    const data = `${"a".repeat(65_535)}😀b`;
+    const send: AtomCommand<TerminalInputTarget, void, never> = {
+      label: "send",
+      run: async (_registry, input) => {
+        sent.push(input.input.data);
+        return AsyncResult.success(undefined);
+      },
+    };
+    const input = createTerminalInputCommand(send);
+    const registry = AtomRegistry.make();
+
+    const result = await input.run(registry, target(data));
+
+    expect(sent.map((chunk) => chunk.length)).toEqual([65_535, 3]);
+    expect(sent.join("")).toBe(data);
+    expect(result._tag).toBe("Success");
+    registry.dispose();
+  });
+
+  it("preserves surrounding input order when splitting an oversized entry", async () => {
+    const sent: string[] = [];
+    const send: AtomCommand<TerminalInputTarget, void, never> = {
+      label: "send",
+      run: async (_registry, input) => {
+        sent.push(input.input.data);
+        return AsyncResult.success(undefined);
+      },
+    };
+    const input = createTerminalInputCommand(send, { maxBatchChars: 4 });
+    const registry = AtomRegistry.make();
+
+    await Promise.all([
+      input.run(registry, target("ab")),
+      input.run(registry, target("cdefgh")),
+      input.run(registry, target("ij")),
+    ]);
+
+    expect(sent).toEqual(["ab", "cdef", "gh", "ij"]);
+    expect(sent.join("")).toBe("abcdefghij");
+    registry.dispose();
+  });
+
+  it("reports a failed chunk after sending the rest of an oversized input", async () => {
+    const sent: string[] = [];
+    const send: AtomCommand<TerminalInputTarget, void, string> = {
+      label: "send",
+      run: async (_registry, input) => {
+        sent.push(input.input.data);
+        return input.input.data === "efgh"
+          ? AsyncResult.failure(Cause.fail("chunk failed"))
+          : AsyncResult.success(undefined);
+      },
+    };
+    const input = createTerminalInputCommand(send, { maxBatchChars: 4 });
+    const registry = AtomRegistry.make();
+
+    const result = await input.run(registry, target("abcdefghij"));
+
+    expect(sent).toEqual(["abcd", "efgh", "ij"]);
+    expect(result._tag).toBe("Failure");
+    registry.dispose();
+  });
+
+  it("fails new input when the bounded lane is full", async () => {
+    let releaseSend!: () => void;
+    const blocked = new Promise<void>((resolve) => {
+      releaseSend = resolve;
+    });
+    const send: AtomCommand<TerminalInputTarget, void, never> = {
+      label: "send",
+      run: async () => {
+        await blocked;
+        return AsyncResult.success(undefined);
+      },
+    };
+    const input = createTerminalInputCommand(send, { maxBufferedChars: 2 });
+    const registry = AtomRegistry.make();
+
+    const accepted = input.run(registry, target("ab"));
+    await Promise.resolve();
+    const rejected = await input.run(registry, target("c"));
+
+    expect(rejected._tag).toBe("Failure");
+    if (rejected._tag === "Failure") {
+      expect(rejected.cause.reasons[0]).toMatchObject({
+        _tag: "Fail",
+        error: expect.any(TerminalInputBackpressureError),
+      });
+    }
+    releaseSend();
+    await accepted;
+    registry.dispose();
+  });
+});