3 &&
+ regularImages.length % 2 === 1 &&
+ index === regularImages.length - 1 &&
+ "col-span-2",
+ )
+ : "aspect-[4/3] overflow-hidden rounded-lg border border-border/80",
)}
>
{image.previewUrl ? (
@@ -535,11 +588,23 @@ function UserTimelineRow({ row }: { row: Extract
) : (
-
+
{image.name}
)}
diff --git a/apps/web/src/components/device/DeviceControlsRail.tsx b/apps/web/src/components/device/DeviceControlsRail.tsx
index 3ed2b3a4fa544..740c342b2cf9f 100644
--- a/apps/web/src/components/device/DeviceControlsRail.tsx
+++ b/apps/web/src/components/device/DeviceControlsRail.tsx
@@ -1,4 +1,4 @@
-import type { DevicePlatform } from "@t3tools/contracts";
+import type { DeviceHostSummary, DevicePlatform } from "@t3tools/contracts";
import {
Camera,
ChevronLeft,
@@ -36,6 +36,7 @@ import type { DeviceControls } from "./useDeviceControls";
export function DeviceControlsRail(props: {
platform: DevicePlatform;
+ hostKind: DeviceHostSummary["kind"] | undefined;
handle: DeviceStreamHandle | null;
view: DeviceViewControls;
controls: DeviceControls;
@@ -46,13 +47,12 @@ export function DeviceControlsRail(props: {
onFloat: () => void;
onClose: () => void;
onPowerOff: () => void;
- /** The host takes input but runs no device commands (Simlock): no rotation, settings, tools, or screenshots. */
- inputOnly?: boolean;
}) {
const { view, handle, controls } = props;
const popupSide = "left";
const settings = controls.detail?.settings;
const inputDisabled = !handle?.inputConnected;
+ const runsHostCommands = props.hostKind === "local" || props.hostKind === "ssh";
const nextAppearance = settings?.appearance === "dark" ? "light" : "dark";
return (
) : (
- Permissions not checked. Connect to this environment to view this session’s permissions.
+ {sessionPermissionsUnavailableMessage({
+ connected,
+ failed: session.hasError,
+ rejected: !session.hasError && !session.isPending && session.data?.authenticated === false,
+ })}
);
}
diff --git a/apps/web/src/lib/turnDiffTree.ts b/apps/web/src/lib/turnDiffTree.ts
index 810a04755e895..e841e18ce24d2 100644
--- a/apps/web/src/lib/turnDiffTree.ts
+++ b/apps/web/src/lib/turnDiffTree.ts
@@ -98,20 +98,6 @@ function toTreeNodes(directory: MutableDirectoryNode): TurnDiffTreeNode[] {
return [...subdirectories, ...files];
}
-export function summarizeTurnDiffStats(files: ReadonlyArray
): TurnDiffStat {
- return files.reduce(
- (acc, file) => {
- const stat = readStat(file);
- if (!stat) return acc;
- return {
- additions: acc.additions + stat.additions,
- deletions: acc.deletions + stat.deletions,
- };
- },
- { additions: 0, deletions: 0 },
- );
-}
-
export function buildTurnDiffTree(files: ReadonlyArray): TurnDiffTreeNode[] {
const root: MutableDirectoryNode = {
name: "",
diff --git a/docs/internals/devices.md b/docs/internals/devices.md
index 5273f7942e9e0..28fa103feb5cb 100644
--- a/docs/internals/devices.md
+++ b/docs/internals/devices.md
@@ -36,13 +36,13 @@ its sessions. Sessions it starts are recorded under the state directory and
stopped by the next server if this one dies; sessions other tools started on
the same account are never touched.
-Each session brings its own serve-sim preview host and agent-device daemon,
-with their own tokens. The proxy asks the host to route each request
-(`DeviceHostReady.upstream`) and adds the session token server-side, so
+Each session brings its own preview host (serve-sim, or serve-emu for Android)
+and agent-device daemon, with their own tokens. The proxy asks the host to
+route each request (`DeviceHostReady.upstream`) and adds the session token server-side, so
clients never hold it. Agent config files are written per device
(`DeviceOperations.agentEndpoint`), which is why `device_open` opens the
-device before targeting it. Device settings run `simctl` on the simulator's
-machine, which EAS does not expose, so they are unavailable there.
+device before targeting it. Device settings and tools run `simctl` or `adb` on
+the device's machine, which EAS does not expose, so they are unavailable there.
## Simlock hosts lease through the agent-device front
@@ -147,8 +147,8 @@ decoder then retains frames for seconds despite `optimizeForLatency` and an empt
`decodeQueueSize`. The iOS AVCC viewer restores the missing metadata before
configuring the decoder so hardware decoding can remain enabled. WebCodecs is
secure-context only, so the viewer probes `isConfigSupported` and falls back to
-MJPEG on iOS when decoding is unavailable. Android has no MJPEG; there the panel
-reports that it cannot decode.
+MJPEG on iOS when decoding is unavailable. Android's SEMU socket has no MJPEG;
+there the panel reports that it cannot decode.
serve-sim can also be locked to WebRTC, as on EAS Simulator, and then refuses
AVCC with a 409 naming the transport. The viewer still asks for AVCC first, so
@@ -156,3 +156,6 @@ hubs that serve it pay nothing, and switches on that 409, or when its MJPEG
image fails in a browser that has `RTCPeerConnection`. Offers are not trickled:
serve-sim takes one offer with every candidate. The video element only feeds
the same canvas sink, so the panel and the mobile WebView keep one surface.
+Android asks serve-emu's `/api` before opening its video socket and streams
+WebRTC when it is advertised (EAS Simulator), keeping a separate `video=0`
+socket for input so input can reconnect without restarting video.
diff --git a/docs/user/background-service.md b/docs/user/background-service.md
index be4ad66ff3ab4..f9ab85fcac4c3 100644
--- a/docs/user/background-service.md
+++ b/docs/user/background-service.md
@@ -90,6 +90,24 @@ Windows background services are not supported.
T3 Connect can offer service installation during setup, but the two are managed
separately. Signing out of T3 Connect does not stop or uninstall the service.
+## Keep the server responsive when agents use all memory
+
+On Linux, agents already run with lower priority than the server. When they
+fill the machine's memory, though, the kernel may still slow down or stop the
+server, because only an administrator can reserve memory for it. Run, as your
+normal user:
+
+```sh
+t3 service protect
+```
+
+It shows the system settings it will change, asks before applying them, and
+then asks for your sudo password. If the machine runs Tailscale or SSH, it
+also offers to protect them, so you can still reach the machine. Memory
+protection takes effect at once. The server's protection from the kernel's
+out-of-memory killer takes effect after `t3 service restart`. `t3 service status` reports whether
+the protection is in place, and `t3 service protect --undo` removes it.
+
## Troubleshooting
Start with `t3 service status` on the host. It prints the log path and, on Linux,
diff --git a/docs/user/devices.md b/docs/user/devices.md
index ce5a7d8d06d05..46a5a427cc87c 100644
--- a/docs/user/devices.md
+++ b/docs/user/devices.md
@@ -94,7 +94,7 @@ started from then on; your own Device panel is unaffected.
The device stream goes through the environment server, so it works over the
local network, Tailscale, and T3 Connect. Live video needs a secure page
(HTTPS or localhost); on a plain-HTTP remote origin iOS falls back to a slower
-still-image stream and Android cannot show video.
+still-image stream and Android cannot show video, except on EAS Simulator.
## SSH device hosts
diff --git a/packages/client-runtime/src/connection/driver.ts b/packages/client-runtime/src/connection/driver.ts
index 0fd08c63364e7..2cdf012187efb 100644
--- a/packages/client-runtime/src/connection/driver.ts
+++ b/packages/client-runtime/src/connection/driver.ts
@@ -40,6 +40,8 @@ export interface EnvironmentConnectionLease {
}
export type RouteCheck = "answered" | "silent" | "unchecked";
+/** "unprepared" means the route answered its check but could not be prepared. */
+export type RoutePreflight = "ready" | "unprepared" | "silent" | "unchecked";
const ROUTE_CHECK_TIMEOUT_MS = 2_500;
// Preferred direct routes share one head start, so a relay connection does not
@@ -57,7 +59,7 @@ export class ConnectionDriver extends Context.Service<
readonly preflight: (
entry: ConnectionCatalogEntry,
route: ConnectionRoute,
- ) => Effect.Effect;
+ ) => Effect.Effect;
/** Asks the host behind a live session to answer a fresh request on the session's own origin. */
readonly checkHost: (prepared: PreparedConnection) => Effect.Effect;
}
@@ -204,13 +206,13 @@ export const make = Effect.gen(function* () {
? Effect.succeed("silent" as const)
: checkRoute(entry, route)
).pipe(
- Effect.flatMap((check) =>
- check === "answered"
- ? resolver.prepare(routeEntry(entry, route)).pipe(
- Effect.as(true),
- Effect.orElseSucceed(() => false),
- )
- : Effect.succeed(false),
+ Effect.filterOrElse(
+ (check): check is Exclude => check !== "answered",
+ () =>
+ resolver.prepare(routeEntry(entry, route)).pipe(
+ Effect.as("ready"),
+ Effect.orElseSucceed((): RoutePreflight => "unprepared"),
+ ),
),
Effect.withSpan("ConnectionDriver.preflight"),
);
diff --git a/packages/client-runtime/src/connection/registry.test.ts b/packages/client-runtime/src/connection/registry.test.ts
index c99e950200b23..ec8863949cef7 100644
--- a/packages/client-runtime/src/connection/registry.test.ts
+++ b/packages/client-runtime/src/connection/registry.test.ts
@@ -431,7 +431,11 @@ const makeHarness = Effect.fn("TestEnvironmentRegistry.makeHarness")(function* (
),
),
preflight: (_entry, route) =>
- checkRoute(route).pipe(Effect.map((check) => check === "answered")),
+ checkRoute(route).pipe(
+ Effect.map((check): ConnectionDriver.RoutePreflight =>
+ check === "answered" ? "ready" : check,
+ ),
+ ),
checkHost: () => Effect.succeed("unchecked"),
});
diff --git a/packages/client-runtime/src/connection/supervisor.test.ts b/packages/client-runtime/src/connection/supervisor.test.ts
index 347bde9cc44c8..66d151043b115 100644
--- a/packages/client-runtime/src/connection/supervisor.test.ts
+++ b/packages/client-runtime/src/connection/supervisor.test.ts
@@ -148,6 +148,7 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?:
readonly initialConfig?: (attempt: number) => Effect.Effect;
readonly checkRoute?: (route: ConnectionRoute) => Effect.Effect;
readonly checkHost?: (prepared: PreparedConnection) => Effect.Effect;
+ readonly preflight?: (route: ConnectionRoute) => Effect.Effect;
readonly inboundFrames?: Effect.Effect;
}) {
const networkStatus = yield* SubscriptionRef.make(
@@ -244,7 +245,12 @@ const makeHarness = Effect.fn("TestConnectionHarness.make")(function* (options?:
ConnectionDriver.ConnectionDriver.of({
connect,
preflight: (_entry, route) =>
- checkRoute(route).pipe(Effect.map((check) => check === "answered")),
+ options?.preflight?.(route) ??
+ checkRoute(route).pipe(
+ Effect.map((check): ConnectionDriver.RoutePreflight =>
+ check === "answered" ? "ready" : check,
+ ),
+ ),
checkHost: (prepared) =>
options?.checkHost?.(prepared) ??
Effect.succeed("unchecked"),
@@ -1614,6 +1620,133 @@ describe("EnvironmentSupervisor routes", () => {
}),
);
+ it.effect("checks a dead LAN route less often until the network changes", () =>
+ Effect.gen(function* () {
+ const lanReachable = yield* Ref.make(false);
+ const lanChecks = yield* Ref.make(0);
+ const harness = yield* makeHarness({
+ checkRoute: (route) =>
+ route.target._tag === "BearerConnectionTarget"
+ ? Ref.update(lanChecks, (count) => count + 1).pipe(
+ Effect.andThen(Ref.get(lanReachable)),
+ Effect.map((reachable) => (reachable ? "answered" : "silent")),
+ )
+ : Effect.succeed("unchecked"),
+ prepare: (_attempt, target) => Effect.succeed(preparedFor(target)),
+ });
+ const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, {
+ initiallyDesired: true,
+ }).pipe(Effect.provide(harness.dependencies));
+
+ yield* awaitState(
+ supervisor.state,
+ (state) => state.phase === "connected" && state.generation === 1,
+ );
+ yield* Ref.set(lanChecks, 0);
+
+ // Checked at 1, 3 and 7 minutes instead of every minute.
+ for (let minute = 0; minute < 10; minute += 1) {
+ yield* TestClock.adjust("60 seconds");
+ }
+ expect(yield* Ref.get(lanChecks)).toBe(3);
+
+ yield* Ref.set(lanReachable, true);
+ yield* harness.wake("network-changed");
+ yield* awaitState(
+ supervisor.state,
+ (state) => state.phase === "connected" && state.generation === 2,
+ );
+ expect(Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target._tag).toBe(
+ "BearerConnectionTarget",
+ );
+ }),
+ );
+
+ it.effect("keeps checking a LAN route every minute when it answers but cannot be prepared", () =>
+ Effect.gen(function* () {
+ const lanPreflights = yield* Ref.make(0);
+ const harness = yield* makeHarness({
+ checkRoute: (route) =>
+ Effect.succeed(route.target._tag === "BearerConnectionTarget" ? "silent" : "unchecked"),
+ preflight: (route) =>
+ route.target._tag === "BearerConnectionTarget"
+ ? Ref.update(lanPreflights, (count) => count + 1).pipe(Effect.as("unprepared" as const))
+ : Effect.succeed("unchecked"),
+ prepare: (_attempt, target) => Effect.succeed(preparedFor(target)),
+ });
+ const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, {
+ initiallyDesired: true,
+ }).pipe(Effect.provide(harness.dependencies));
+
+ yield* awaitState(
+ supervisor.state,
+ (state) => state.phase === "connected" && state.generation === 1,
+ );
+ for (let minute = 0; minute < 10; minute += 1) {
+ yield* TestClock.adjust("60 seconds");
+ }
+ expect(yield* Ref.get(lanPreflights)).toBe(10);
+ }),
+ );
+
+ it.effect("ignores a dead LAN check that was running when the network changed", () =>
+ Effect.gen(function* () {
+ const lanChecks = yield* Ref.make(0);
+ const checkStarted = yield* Deferred.make();
+ const finishCheck = yield* Deferred.make();
+ const woke = yield* Ref.make(false);
+ const wakeupHandled = yield* Deferred.make();
+ const harness = yield* makeHarness({
+ checkRoute: (route) =>
+ route.target._tag === "BearerConnectionTarget"
+ ? Ref.updateAndGet(lanChecks, (count) => count + 1).pipe(
+ Effect.flatMap((check) =>
+ check === 1
+ ? Effect.succeed("silent" as const)
+ : check === 2
+ ? Deferred.succeed(checkStarted, undefined).pipe(
+ Effect.andThen(Deferred.await(finishCheck)),
+ Effect.as("silent" as const),
+ )
+ : Effect.succeed("answered" as const),
+ ),
+ )
+ : Effect.succeed("unchecked"),
+ prepare: (_attempt, target) => Effect.succeed(preparedFor(target)),
+ // The wakeup probes the session only after it forgets the silent routes.
+ probe: () =>
+ Ref.get(woke).pipe(
+ Effect.flatMap((afterWake) =>
+ afterWake ? Deferred.succeed(wakeupHandled, undefined) : Effect.void,
+ ),
+ Effect.asVoid,
+ ),
+ });
+ const supervisor = yield* EnvironmentSupervisor.make(LAN_THEN_RELAY_ENTRY, {
+ initiallyDesired: true,
+ }).pipe(Effect.provide(harness.dependencies));
+
+ yield* awaitState(
+ supervisor.state,
+ (state) => state.phase === "connected" && state.generation === 1,
+ );
+ yield* TestClock.adjust("60 seconds");
+ yield* Deferred.await(checkStarted);
+ yield* Ref.set(woke, true);
+ yield* harness.wake("network-changed");
+ yield* Deferred.await(wakeupHandled);
+ yield* Deferred.succeed(finishCheck, undefined);
+
+ yield* awaitState(
+ supervisor.state,
+ (state) => state.phase === "connected" && state.generation === 2,
+ );
+ expect(Option.getOrThrow(yield* SubscriptionRef.get(supervisor.prepared)).target._tag).toBe(
+ "BearerConnectionTarget",
+ );
+ }),
+ );
+
it.effect("moves off a dead LAN socket when the phone switches to cellular", () =>
Effect.gen(function* () {
const lanReachable = yield* Ref.make(true);
diff --git a/packages/client-runtime/src/connection/supervisor.ts b/packages/client-runtime/src/connection/supervisor.ts
index ce7632df0f52b..212381921961b 100644
--- a/packages/client-runtime/src/connection/supervisor.ts
+++ b/packages/client-runtime/src/connection/supervisor.ts
@@ -58,7 +58,11 @@ const SILENT_SOCKET_RECHECK_EVERY = "1 second";
// instead of timing out.
const REMOTE_SILENT_SOCKET_AFTER = "2500 millis";
const BACKOFF_RESET_AFTER_MS = 30_000;
-const BETTER_ROUTE_CHECK_INTERVAL = "60 seconds";
+const BETTER_ROUTE_CHECK_INTERVAL_MS = 60_000;
+// Each check of a dead route can wait out the full probe timeout, so a route
+// that keeps failing is checked half as often each time, until the network
+// changes and it may answer again.
+const SILENT_ROUTE_BACKOFF_MAX_MS = 1_800_000;
// A better route that answered the check but then failed to connect is not
// tried again for this long, so a flaky LAN cannot bounce the connection.
const BETTER_ROUTE_COOLDOWN_MS = 300_000;
@@ -139,6 +143,20 @@ function retryDelayMs(failureCount: number, random: number): number {
return Math.round(ceiling / 2 + (ceiling / 2) * random);
}
+interface SilentRoute {
+ readonly failures: number;
+ readonly until: bigint;
+}
+
+function backOffSilentRoute(previous: SilentRoute | undefined, now: bigint): SilentRoute {
+ const failures = (previous?.failures ?? 0) + 1;
+ const delayMs = Math.min(
+ SILENT_ROUTE_BACKOFF_MAX_MS,
+ BETTER_ROUTE_CHECK_INTERVAL_MS * 2 ** failures,
+ );
+ return { failures, until: now + BigInt(delayMs) * 1_000_000n };
+}
+
function annotateTarget(target: ConnectionTarget) {
return Effect.annotateCurrentSpan({
"environment.id": target.environmentId,
@@ -288,6 +306,16 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
const currentRoutes = Ref.get(currentEntry).pipe(Effect.map(connectionRoutes));
const preferredRouteId = yield* Ref.make(Option.none());
const routeCooldowns = yield* Ref.make>(new Map());
+ // The generation keeps a check that started before a reset from recording
+ // its stale failures after it.
+ const silentRoutes = yield* Ref.make({
+ generation: 0,
+ routes: new Map() as ReadonlyMap,
+ });
+ const forgetSilentRoutes = Ref.update(silentRoutes, ({ generation }) => ({
+ generation: generation + 1,
+ routes: new Map(),
+ }));
const attemptEntry = Effect.gen(function* () {
const latest = yield* Ref.get(currentEntry);
@@ -338,15 +366,29 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
if (current <= 0) return;
const now = yield* Clock.monotonicTimeNanos;
const cooldowns = yield* Ref.get(routeCooldowns);
+ const silent = yield* Ref.get(silentRoutes);
const better = connectionRoutes(latest)
.slice(0, current)
- .filter((route) => (cooldowns.get(connectionRouteId(route.target)) ?? 0n) <= now);
- const passed = yield* Effect.forEach(
+ .filter((route) => {
+ const id = connectionRouteId(route.target);
+ return (cooldowns.get(id) ?? 0n) <= now && (silent.routes.get(id)?.until ?? 0n) <= now;
+ });
+ const results = yield* Effect.forEach(
better,
(route: ConnectionRoute) => driver.preflight(latest, route),
{ concurrency: "unbounded" },
);
- const index = passed.indexOf(true);
+ yield* Ref.update(silentRoutes, (previous) => {
+ if (previous.generation !== silent.generation) return previous;
+ const routes = new Map(previous.routes);
+ for (const [position, route] of better.entries()) {
+ const id = connectionRouteId(route.target);
+ if (results[position] !== "silent") routes.delete(id);
+ else routes.set(id, backOffSilentRoute(previous.routes.get(id), now));
+ }
+ return { generation: previous.generation, routes };
+ });
+ const index = results.indexOf("ready");
if (index === -1) return;
const live = yield* SubscriptionRef.get(session);
if (Option.isNone(live) || live.value !== lease.session) return;
@@ -668,7 +710,7 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
);
// Ticks always run: the check skips the preferred route itself, and the
// route list can grow while connected (learned routes).
- yield* Stream.tick(BETTER_ROUTE_CHECK_INTERVAL).pipe(
+ yield* Stream.tick(Duration.millis(BETTER_ROUTE_CHECK_INTERVAL_MS)).pipe(
Stream.drop(1),
Stream.runForEach(() => requestBetterRouteCheck(lease)),
Effect.forkScoped,
@@ -1114,14 +1156,24 @@ export const make = Effect.fn("EnvironmentSupervisor.make")(function* (
current.network === network ? [false, current] : ([true, { ...current, network }] as const),
).pipe(
Effect.flatMap((changed) =>
- changed ? signal({ _tag: "NetworkChanged", network }) : Effect.void,
+ changed
+ ? (network === "offline" ? Effect.void : forgetSilentRoutes).pipe(
+ Effect.andThen(signal({ _tag: "NetworkChanged", network })),
+ )
+ : Effect.void,
),
),
),
Effect.forkScoped,
);
yield* wakeups.changes.pipe(
- Stream.runForEach((reason) => signal({ _tag: "Wakeup", reason })),
+ Stream.runForEach((reason) =>
+ // A long suspension can hide a network change the platform never reported.
+ (reason === "network-changed" || reason === "application-active-long"
+ ? forgetSilentRoutes
+ : Effect.void
+ ).pipe(Effect.andThen(signal({ _tag: "Wakeup", reason }))),
+ ),
Effect.forkScoped,
);
yield* Queue.take(betterRouteChecks).pipe(
diff --git a/packages/client-runtime/src/device/stream.test.ts b/packages/client-runtime/src/device/stream.test.ts
index 737daf5158c04..09f04c52780e2 100644
--- a/packages/client-runtime/src/device/stream.test.ts
+++ b/packages/client-runtime/src/device/stream.test.ts
@@ -75,7 +75,7 @@ describe("native device stream transport", () => {
}
}
vi.stubGlobal("WebSocket", FakeSocket);
- const fetch = vi.fn(() => Promise.resolve(new Response("frame")));
+ const fetch = vi.fn(() => Promise.resolve(Response.json({})));
vi.stubGlobal("fetch", fetch);
const events = {
onStatus: vi.fn(),
@@ -251,6 +251,7 @@ function recoveryFixture(
const fetch = vi.fn((url: string, init: RequestInit) => {
const signal = init.signal!;
signals.push(signal);
+ if (url.includes("/api?")) return Promise.resolve(Response.json({}));
if (!url.includes("stream.avcc")) return Promise.resolve(new Response("prime"));
return Promise.resolve(
new Response(
@@ -450,9 +451,11 @@ describe("shared device stream readiness and recovery", () => {
}),
);
client.start();
+ await vi.advanceTimersByTimeAsync(0);
sockets[0]!.onmessage?.({ data: sps });
client.stop();
client.start();
+ await vi.advanceTimersByTimeAsync(0);
resolveSupport({ supported: true });
await vi.advanceTimersByTimeAsync(0);
expect(decoders).toHaveLength(0);
@@ -491,3 +494,136 @@ describe("shared device stream readiness and recovery", () => {
expect(vi.getTimerCount()).toBe(0);
});
});
+
+describe("Android WebRTC device streams", () => {
+ afterEach(() => {
+ vi.useRealTimers();
+ vi.unstubAllGlobals();
+ });
+
+ function setup() {
+ vi.useFakeTimers();
+ const sockets: Socket[] = [];
+ class Socket {
+ static OPEN = 1;
+ readyState = 1;
+ binaryType = "";
+ onopen: (() => void) | null = null;
+ onmessage: ((event: { data: string }) => void) | null = null;
+ onclose: ((event: { code: number; reason: string }) => void) | null = null;
+ send = vi.fn();
+ close = vi.fn();
+ readonly url: string;
+ constructor(url: string) {
+ this.url = url;
+ sockets.push(this);
+ }
+ }
+ const videos: Video[] = [];
+ class Video {
+ muted = false;
+ playsInline = false;
+ style = { cssText: "" };
+ srcObject: unknown = null;
+ videoWidth = 400;
+ videoHeight = 800;
+ setAttribute = vi.fn();
+ remove = vi.fn();
+ play = vi.fn(async () => {});
+ callback: (() => void) | null = null;
+ requestVideoFrameCallback(callback: () => void) {
+ this.callback = callback;
+ return 0;
+ }
+ constructor() {
+ videos.push(this);
+ }
+ }
+ const peers: Peer[] = [];
+ class Peer {
+ iceGatheringState = "complete";
+ connectionState = "connected";
+ localDescription = { type: "offer", sdp: "offer-sdp" };
+ ontrack: ((event: { streams: unknown[]; track: unknown }) => void) | null = null;
+ onconnectionstatechange: (() => void) | null = null;
+ close = vi.fn();
+ createOffer = vi.fn(async () => this.localDescription);
+ setLocalDescription = vi.fn(async () => {});
+ setRemoteDescription = vi.fn(async () => {});
+ addTransceiver = vi.fn(() => ({ setCodecPreferences: vi.fn() }));
+ addEventListener = vi.fn();
+ removeEventListener = vi.fn();
+ constructor() {
+ peers.push(this);
+ }
+ }
+ vi.stubGlobal("WebSocket", Socket);
+ vi.stubGlobal("RTCPeerConnection", Peer);
+ vi.stubGlobal("RTCRtpReceiver", { getCapabilities: () => ({ codecs: [] }) });
+ vi.stubGlobal("document", { createElement: () => new Video(), body: { append: vi.fn() } });
+ const fetch = vi.fn((url: string, _init?: RequestInit) =>
+ Promise.resolve(
+ url.includes("/api?")
+ ? Response.json({ serial: "emulator-5554", stream: { transport: "webrtc" } })
+ : Response.json({ type: "answer", sdp: "answer-sdp" }),
+ ),
+ );
+ vi.stubGlobal("fetch", fetch);
+ const events = {
+ onStatus: vi.fn(),
+ onScreen: vi.fn(),
+ onUnauthorized: vi.fn(),
+ onMjpegFallback: vi.fn(),
+ onInputConnected: vi.fn(),
+ };
+ const present = vi.fn(() => true);
+ const client = createDeviceStreamClient(
+ {
+ platform: "android",
+ deviceId: "session-1",
+ access: {
+ httpBase: "https://test/api/device-hub",
+ wsBase: "wss://test/api/device-hub",
+ credentials: true,
+ query: { hostId: "eas" },
+ },
+ },
+ { present },
+ events,
+ );
+ return { client, events, fetch, peers, videos, sockets, present };
+ }
+
+ it("streams over WebRTC when serve-emu advertises it, with input that reconnects on its own", async () => {
+ const { client, events, fetch, peers, videos, sockets, present } = setup();
+ client.start();
+ await vi.advanceTimersByTimeAsync(0);
+ expect(fetch.mock.calls.map(([url]) => url)).toEqual([
+ "https://test/api/device-hub/vendor/serve-emu/api?device=session-1&hostId=eas",
+ "https://test/api/device-hub/vendor/serve-emu/webrtc/offer?device=session-1&hostId=eas",
+ ]);
+ expect(peers[0]?.setRemoteDescription).toHaveBeenCalledWith({
+ type: "answer",
+ sdp: "answer-sdp",
+ });
+ peers[0]?.ontrack?.({ streams: [{}], track: {} });
+ videos[0]?.callback?.();
+ expect(present).toHaveBeenCalledWith(videos[0], 400, 800);
+ expect(events.onStatus).toHaveBeenLastCalledWith("streaming", undefined);
+
+ expect(sockets.map((socket) => socket.url)).toEqual([
+ "wss://test/api/device-hub/vendor/serve-emu/ws?device=session-1&video=0&hostId=eas",
+ ]);
+ sockets[0]!.onopen?.();
+ client.pressButton("back");
+ expect(sockets[0]!.send).toHaveBeenCalledWith(JSON.stringify({ type: "back" }));
+ events.onStatus.mockClear();
+ sockets[0]!.onclose?.({ code: 1001, reason: "going away" });
+ await vi.advanceTimersByTimeAsync(1_000);
+ expect(sockets[1]?.url).toContain("video=0");
+ expect(peers).toHaveLength(1);
+ expect(peers[0]?.close).not.toHaveBeenCalled();
+ expect(events.onStatus).not.toHaveBeenCalled();
+ client.stop();
+ });
+});
diff --git a/packages/client-runtime/src/device/stream.ts b/packages/client-runtime/src/device/stream.ts
index c4b8078597245..a8913d5a41037 100644
--- a/packages/client-runtime/src/device/stream.ts
+++ b/packages/client-runtime/src/device/stream.ts
@@ -836,9 +836,10 @@ export function createDeviceStreamClient(
/**
* serve-sim can be locked to WebRTC (EAS Simulator always is), and then
- * refuses AVCC and MJPEG. An unreadable answer keeps the AVCC path.
+ * refuses AVCC and MJPEG; serve-emu advertises WebRTC the same way. An
+ * unreadable answer keeps the default path.
*/
- const probeIosTransport = async (): Promise => {
+ const probeWebRtc = async (): Promise => {
const probe = new AbortController();
const timeout = setTimeout(() => probe.abort(), TRANSPORT_PROBE_TIMEOUT_MS);
try {
@@ -858,6 +859,9 @@ export function createDeviceStreamClient(
}
};
+ const webRtcUrl = (action: "offer" | "close") =>
+ httpUrl(avcc ? `/helper/${device}/webrtc/${action}` : `/webrtc/${action}?device=${device}`);
+
const closePeer = () => {
const current = peer;
if (!current) return;
@@ -865,8 +869,8 @@ export function createDeviceStreamClient(
current.connection.close();
current.video.srcObject = null;
current.video.remove();
- // Releases serve-sim's capture session; best effort because the page may be unloading.
- void fetch(httpUrl(`/helper/${device}/webrtc/close`), {
+ // Releases the capture session; best effort because the page may be unloading.
+ void fetch(webRtcUrl("close"), {
method: "POST",
keepalive: true,
headers: { "content-type": "application/json" },
@@ -875,7 +879,7 @@ export function createDeviceStreamClient(
}).catch(() => {});
};
- const readIosWebRtc = async (settings: WebRtcSettings) => {
+ const readWebRtc = async (settings: WebRtcSettings) => {
const lifecycle = generation;
const feedGeneration = ++videoGeneration;
closePeer();
@@ -920,7 +924,7 @@ export function createDeviceStreamClient(
if (!isCurrent()) return;
closePeer();
connecting(detail);
- scheduleRetry("video", () => void readIosWebRtc(settings));
+ scheduleRetry("video", () => void readWebRtc(settings));
};
const presentFrame = () => {
if (!isCurrent()) {
@@ -988,7 +992,7 @@ export function createDeviceStreamClient(
iceGatheringDeadlineMs = Math.min(iceGatheringDeadlineMs * 2, ICE_GATHERING_MAX_TIMEOUT_MS);
return retry("Device stream found no network candidates yet. Retrying…");
}
- const response = await fetch(httpUrl(`/helper/${device}/webrtc/offer`), {
+ const response = await fetch(webRtcUrl("offer"), {
method: "POST",
headers: { "content-type": "application/json" },
body: JSON.stringify({
@@ -1023,8 +1027,7 @@ export function createDeviceStreamClient(
const switchToWebRtc = async (failure: string, locked: boolean) => {
const session = generation;
if (typeof RTCPeerConnection === "undefined") return fail(failure);
- const settings =
- (await probeIosTransport()) ?? (locked ? { transport: "webrtc" as const } : null);
+ const settings = (await probeWebRtc()) ?? (locked ? { transport: "webrtc" as const } : null);
if (stopped || generation !== session) return;
if (!settings) return fail(failure);
if (mjpeg) {
@@ -1036,7 +1039,7 @@ export function createDeviceStreamClient(
// MJPEG may have shown frames already: say the stream is connecting again
// rather than leave a blank canvas until the first WebRTC frame.
connecting("Switching to WebRTC.");
- void readIosWebRtc(settings);
+ void readWebRtc(settings);
};
const startDuoVideo = (panels: DuoPanelSinks) => {
@@ -1161,18 +1164,21 @@ export function createDeviceStreamClient(
ws.onerror = () => ws.close();
};
- const connectAndroid = () => {
+ /** `inputOnly` keeps a socket for input beside a WebRTC feed, so its reconnects leave video alone. */
+ const connectAndroid = (inputOnly = false) => {
if (stopped) return;
- const ws = new WebSocket(wsUrl(`/ws?device=${device}&frame-meta=1`));
+ const ws = new WebSocket(
+ wsUrl(`/ws?device=${device}&${inputOnly ? "video=0" : "frame-meta=1"}`),
+ );
ws.binaryType = "arraybuffer";
socket = ws;
ws.onopen = () => {
if (stopped || socket !== ws) return;
- connecting();
+ if (!inputOnly) connecting();
events.onInputConnected(true);
};
ws.onmessage = (event) => {
- if (stopped || socket !== ws) return;
+ if (stopped || socket !== ws || inputOnly) return;
if (typeof event.data === "string") {
// The encoder restarts at a new size when the device rotates; the
// next keyframe carries a fresh SPS, so the decoder is rebuilt from it.
@@ -1224,13 +1230,29 @@ export function createDeviceStreamClient(
(event.code === 1006 && access.query.wsTicket)
)
return handleUnauthorized();
- configuring = false;
- connecting(event.reason || undefined);
- scheduleRetry("input", connectAndroid);
+ if (!inputOnly) {
+ configuring = false;
+ connecting(event.reason || undefined);
+ }
+ // A full socket re-probes, so one failed /api read does not pin SEMU on a WebRTC host.
+ scheduleRetry("input", () => (inputOnly ? connectAndroid(true) : void startAndroid()));
};
ws.onerror = () => ws.close();
};
+ const startAndroid = async () => {
+ const session = generation;
+ const settings = await probeWebRtc();
+ if (stopped || generation !== session) return;
+ if (settings) {
+ if (typeof RTCPeerConnection === "undefined")
+ return fail("This browser cannot receive the device's WebRTC stream.");
+ if (!target.videoOnly) connectAndroid(true);
+ void readWebRtc(settings);
+ } else if (useWebCodecs) connectAndroid();
+ else fail("This browser cannot decode the Android stream (WebCodecs unavailable).");
+ };
+
const start = () => {
if (!stopped) return;
stopped = false;
@@ -1247,11 +1269,7 @@ export function createDeviceStreamClient(
if (!target.videoOnly) void connectIosInput();
if (useWebCodecs) void readIosVideo();
else fallBackToMjpeg();
- } else if (useWebCodecs) {
- connectAndroid();
- } else {
- fail("This browser cannot decode the Android stream (WebCodecs unavailable).");
- }
+ } else void startAndroid();
};
const stop = () => {
diff --git a/packages/client-runtime/src/state/session.test.ts b/packages/client-runtime/src/state/session.test.ts
new file mode 100644
index 0000000000000..740ab2229bdb4
--- /dev/null
+++ b/packages/client-runtime/src/state/session.test.ts
@@ -0,0 +1,133 @@
+import { describe, expect, it } from "@effect/vitest";
+import { EnvironmentId, type AuthSessionState } from "@t3tools/contracts";
+import * as Clock from "effect/Clock";
+import * as Effect from "effect/Effect";
+import * as Layer from "effect/Layer";
+import * as Option from "effect/Option";
+import * as Queue from "effect/Queue";
+import * as Stream from "effect/Stream";
+import * as SubscriptionRef from "effect/SubscriptionRef";
+import * as TestClock from "effect/testing/TestClock";
+import { AsyncResult, Atom, AtomRegistry } from "effect/reactivity";
+
+import {
+ AVAILABLE_CONNECTION_STATE,
+ PrimaryConnectionTarget,
+ type PreparedConnection,
+} from "../connection/model.ts";
+import * as EnvironmentRegistry from "../connection/registry.ts";
+import * as EnvironmentSupervisor from "../connection/supervisor.ts";
+import * as RpcHttp from "../rpc/http.ts";
+import type { RpcSession } from "../rpc/session.ts";
+import { createEnvironmentSessionAtoms } from "./session.ts";
+
+const TARGET = new PrimaryConnectionTarget({
+ environmentId: EnvironmentId.make("environment-1"),
+ label: "Test environment",
+ httpBaseUrl: "https://environment.example.test",
+ wsBaseUrl: "wss://environment.example.test",
+});
+const PREPARED: PreparedConnection = {
+ environmentId: TARGET.environmentId,
+ label: TARGET.label,
+ httpBaseUrl: "https://environment.example.test",
+ socketUrl: "wss://environment.example.test/ws",
+ httpAuthorization: { _tag: "Bearer", token: "bearer-token" },
+ target: TARGET,
+};
+const SESSION = {
+ authenticated: true,
+ auth: {
+ policy: "remote-reachable",
+ bootstrapMethods: [],
+ sessionMethods: [],
+ sessionCookieName: "t3_session",
+ },
+ scopes: ["orchestration:read"],
+} satisfies AuthSessionState;
+
+const rpcSession = (): RpcSession => ({ closed: Effect.never }) as unknown as RpcSession;
+
+const makeHarness = Effect.fn("TestSessionState.makeHarness")(function* (
+ reply: (requestNumber: number) => Promise,
+) {
+ let requestCount = 0;
+ const transport = yield* SubscriptionRef.make(Option.some(rpcSession()));
+ const supervisor = EnvironmentSupervisor.EnvironmentSupervisor.of({
+ target: TARGET,
+ state: yield* SubscriptionRef.make(AVAILABLE_CONNECTION_STATE),
+ session: transport,
+ prepared: yield* SubscriptionRef.make(Option.some(PREPARED)),
+ connect: Effect.void,
+ disconnect: Effect.void,
+ retryNow: Effect.void,
+ } satisfies EnvironmentSupervisor.EnvironmentSupervisor["Service"]);
+ const followStream: EnvironmentRegistry.EnvironmentRegistry["Service"]["followStream"] = (
+ _environmentId,
+ stream,
+ ) => Stream.provideService(stream, EnvironmentSupervisor.EnvironmentSupervisor, supervisor);
+ const runtime = Atom.runtime(
+ Layer.mergeAll(
+ Layer.succeed(
+ EnvironmentRegistry.EnvironmentRegistry,
+ EnvironmentRegistry.EnvironmentRegistry.of({
+ followStream,
+ } as unknown as EnvironmentRegistry.EnvironmentRegistry["Service"]),
+ ),
+ RpcHttp.layerRemoteHttpClient(async () => reply(++requestCount)),
+ // The test's clock, so the retry wait moves with `TestClock.adjust`.
+ Layer.succeed(Clock.Clock, yield* Clock.Clock),
+ ),
+ );
+ const registry = yield* Effect.acquireRelease(Effect.sync(AtomRegistry.make), (registry) =>
+ Effect.sync(() => registry.dispose()),
+ );
+ const results = yield* Queue.unbounded>();
+ const atom = createEnvironmentSessionAtoms(runtime).sessionStateAtom(TARGET.environmentId);
+ const unsubscribe = registry.subscribe(atom, (result) => Queue.offerUnsafe(results, result), {
+ immediate: true,
+ });
+ yield* Effect.addFinalizer(() => Effect.sync(unsubscribe));
+ const nextSettled = Effect.fn("TestSessionState.nextSettled")(function* () {
+ for (;;) {
+ const result = yield* Queue.take(results);
+ if (!result.waiting && result._tag !== "Initial") return result;
+ }
+ });
+ return { transport, nextSettled };
+});
+
+const unreachable = () => Promise.reject(new TypeError("Network request failed"));
+
+describe("environment session state", () => {
+ it.effect("fetches permissions again when the connection is re-established", () =>
+ Effect.gen(function* () {
+ const harness = yield* makeHarness((requestNumber) =>
+ requestNumber === 1 ? unreachable() : Promise.resolve(Response.json(SESSION)),
+ );
+ expect((yield* harness.nextSettled())._tag).toBe("Failure");
+
+ yield* SubscriptionRef.set(harness.transport, Option.some(rpcSession()));
+
+ expect(yield* harness.nextSettled()).toMatchObject({ _tag: "Success", value: SESSION });
+ }),
+ );
+
+ it.effect("retries a failed permission check with backoff that restarts per connection", () =>
+ Effect.gen(function* () {
+ const harness = yield* makeHarness((requestNumber) =>
+ requestNumber <= 3 ? unreachable() : Promise.resolve(Response.json(SESSION)),
+ );
+ expect((yield* harness.nextSettled())._tag).toBe("Failure");
+
+ yield* TestClock.adjust("2 seconds");
+ expect((yield* harness.nextSettled())._tag).toBe("Failure");
+
+ yield* SubscriptionRef.set(harness.transport, Option.some(rpcSession()));
+ expect((yield* harness.nextSettled())._tag).toBe("Failure");
+
+ yield* TestClock.adjust("2 seconds");
+ expect(yield* harness.nextSettled()).toMatchObject({ _tag: "Success", value: SESSION });
+ }),
+ );
+});
diff --git a/packages/client-runtime/src/state/session.ts b/packages/client-runtime/src/state/session.ts
index efdf0b37c0a09..90afa29f9d05d 100644
--- a/packages/client-runtime/src/state/session.ts
+++ b/packages/client-runtime/src/state/session.ts
@@ -14,6 +14,7 @@ import type { PreparedConnection } from "../connection/model.ts";
import * as EnvironmentSupervisor from "../connection/supervisor.ts";
import * as ManagedRelay from "../relay/managedRelay.ts";
import type { RemoteEnvironmentRequestError } from "../rpc/http.ts";
+import type { RpcSession } from "../rpc/session.ts";
import { safeErrorLogAttributes } from "../errors/safeLog.ts";
import { executeAuthenticatedEnvironmentHttpRequest } from "./environmentHttpAuth.ts";
import { followStreamInEnvironment } from "./environmentStreams.ts";
@@ -39,6 +40,8 @@ function initialConfigOption(
const DEFAULT_SESSION_STATE_TIMEOUT_MS = 6_000;
const MAX_SESSION_STATE_TIMEOUT_MS = 60_000;
+const SESSION_STATE_RETRY_BASE_MS = 2_000;
+const SESSION_STATE_RETRY_MAX_MS = 60_000;
export const fetchEnvironmentSessionState = Effect.fn(
"clientRuntime.state.fetchEnvironmentSessionState",
@@ -136,13 +139,38 @@ function makeEnvironmentSessionAtoms(
).pipe(Atom.withLabel(`environment-prepared-connection:${environmentId}`)),
);
- const sessionStateAtom = Atom.family((environmentId: EnvironmentId) =>
- runtime
+ const transportSessionAtom = Atom.family((environmentId: EnvironmentId) =>
+ runtime.atom(
+ followStreamInEnvironment(
+ environmentId,
+ Stream.unwrap(
+ EnvironmentSupervisor.EnvironmentSupervisor.pipe(
+ Effect.map((supervisor) => SubscriptionRef.changes(supervisor.session)),
+ ),
+ ),
+ ),
+ { initialValue: Option.none() },
+ ),
+ );
+
+ const sessionStateAtom = Atom.family((environmentId: EnvironmentId) => {
+ let consecutiveFailures = 0;
+ let backoffConnection: RpcSession | null = null;
+ return runtime
.atom((get) => {
const prepared = Option.getOrNull(get(preparedConnectionValueAtom(environmentId)));
- if (prepared === null) {
+ // Each established connection fetches anew, and a failure retries while it stays up,
+ // so a fetch that failed mid-reconnect cannot outlive the socket that recovered.
+ const connection = AsyncResult.getOrElse(get(transportSessionAtom(environmentId)), () =>
+ Option.none(),
+ );
+ if (prepared === null || Option.isNone(connection)) {
return Effect.never;
}
+ if (connection.value !== backoffConnection) {
+ backoffConnection = connection.value;
+ consecutiveFailures = 0;
+ }
return Effect.gen(function* () {
const signer = yield* Effect.serviceOption(ManagedRelay.ManagedRelayDpopSigner);
const remoteAuthorization = yield* Effect.serviceOption(
@@ -155,14 +183,31 @@ function makeEnvironmentSessionAtoms(
signer,
remoteAuthorization,
}).pipe(Effect.provideService(HttpClient.HttpClient, client.value));
- });
+ }).pipe(
+ Effect.tap(() =>
+ Effect.sync(() => {
+ consecutiveFailures = 0;
+ }),
+ ),
+ Effect.tapError(() => {
+ const delayMs = Math.min(
+ SESSION_STATE_RETRY_BASE_MS * 2 ** consecutiveFailures,
+ SESSION_STATE_RETRY_MAX_MS,
+ );
+ consecutiveFailures += 1;
+ return Effect.sleep(delayMs).pipe(
+ Effect.andThen(Effect.sync(() => get.refreshSelf())),
+ Effect.forkScoped,
+ );
+ }),
+ );
})
.pipe(
Atom.swr({ staleTime: 30_000, revalidateOnMount: true }),
Atom.setIdleTTL(5 * 60_000),
Atom.withLabel(`environment-session-state:${environmentId}`),
- ),
- );
+ );
+ });
const sessionStateValueAtom = Atom.family((environmentId: EnvironmentId) =>
Atom.make(
diff --git a/packages/client-runtime/src/state/threadCheckpoints.ts b/packages/client-runtime/src/state/threadCheckpoints.ts
index 4501007e20053..ee2fe45d9c756 100644
--- a/packages/client-runtime/src/state/threadCheckpoints.ts
+++ b/packages/client-runtime/src/state/threadCheckpoints.ts
@@ -3,6 +3,7 @@ import type {
CheckpointRef,
CheckpointScopeId,
MessageId,
+ OrchestrationV2CheckpointFileTotals,
OrchestrationV2ThreadProjection,
RunId,
} from "@t3tools/contracts";
@@ -15,12 +16,14 @@ export interface ThreadCheckpointSummary {
readonly checkpointTurnCount: number;
readonly checkpointRef: CheckpointRef;
readonly status: "ready" | "missing" | "error" | "stale";
+ /** A prefix of the changed files when `filesTotal` is set; see `checkpointFileTotals`. */
readonly files: ReadonlyArray<{
readonly path: string;
readonly kind: string;
readonly additions: number;
readonly deletions: number;
}>;
+ readonly filesTotal?: OrchestrationV2CheckpointFileTotals;
readonly assistantMessageId: MessageId | null;
readonly completedAt: string;
}
@@ -43,9 +46,23 @@ export function deriveThreadCheckpointSummaries(
checkpointRef: checkpoint.ref,
status: checkpoint.status,
files: checkpoint.files,
+ ...(checkpoint.filesTotal === undefined ? {} : { filesTotal: checkpoint.filesTotal }),
assistantMessageId,
completedAt: DateTime.formatIso(checkpoint.capturedAt),
},
];
});
}
+
+export function checkpointFileTotals(
+ checkpoint: Pick,
+): OrchestrationV2CheckpointFileTotals {
+ if (checkpoint.filesTotal !== undefined) return checkpoint.filesTotal;
+ let additions = 0;
+ let deletions = 0;
+ for (const file of checkpoint.files) {
+ additions += file.additions;
+ deletions += file.deletions;
+ }
+ return { count: checkpoint.files.length, additions, deletions };
+}
diff --git a/packages/client-runtime/src/userMessage.test.ts b/packages/client-runtime/src/userMessage.test.ts
index 0447dc54d1b26..207939cf2f39d 100644
--- a/packages/client-runtime/src/userMessage.test.ts
+++ b/packages/client-runtime/src/userMessage.test.ts
@@ -1,6 +1,7 @@
import { describe, expect, it } from "vite-plus/test";
import { GuestMessageId } from "@t3tools/contracts";
+import { ATTACHMENT_ONLY_BOOTSTRAP_PROMPT } from "@t3tools/shared/composerContextReferences";
import { resolveUserMessagePresentation } from "./userMessage.ts";
@@ -21,6 +22,19 @@ describe("resolveUserMessagePresentation", () => {
}
});
+ it("hides the prompt that stands in for an attachment-only message", () => {
+ for (const text of [
+ ATTACHMENT_ONLY_BOOTSTRAP_PROMPT,
+ `Ultrathink:\n${ATTACHMENT_ONLY_BOOTSTRAP_PROMPT}`,
+ ]) {
+ expect(resolveUserMessagePresentation({ role: "user", text })).toEqual({
+ text: "",
+ attribution: null,
+ scheduledTaskId: undefined,
+ });
+ }
+ });
+
it("shows a guest's own words and name, not the attributed text the agent reads", () => {
expect(
resolveUserMessagePresentation({
diff --git a/packages/client-runtime/src/userMessage.ts b/packages/client-runtime/src/userMessage.ts
index d5d7c5ac08956..8a80f31b155d0 100644
--- a/packages/client-runtime/src/userMessage.ts
+++ b/packages/client-runtime/src/userMessage.ts
@@ -5,6 +5,7 @@ import {
type OrchestrationV2IdleCompaction,
ScheduledTaskId,
} from "@t3tools/contracts";
+import { isAttachmentOnlyBootstrapPrompt } from "@t3tools/shared/composerContextReferences";
/** The working row while T3 Code's own compaction runs. */
export const IDLE_COMPACTING_LABELS = {
@@ -68,7 +69,8 @@ export function resolveUserMessagePresentation(message: {
};
}
return {
- text: message.text,
+ // The agent needs words to start a turn; people sent only the attachments.
+ text: isAttachmentOnlyBootstrapPrompt(message.text) ? "" : message.text,
attribution:
message.createdBy !== "agent"
? null
diff --git a/packages/contracts/src/orchestrationV2.ts b/packages/contracts/src/orchestrationV2.ts
index 747d05e7ce4eb..026e74831233d 100644
--- a/packages/contracts/src/orchestrationV2.ts
+++ b/packages/contracts/src/orchestrationV2.ts
@@ -1349,6 +1349,14 @@ export const OrchestrationV2CheckpointFileSummary = Schema.Struct({
});
export type OrchestrationV2CheckpointFileSummary = typeof OrchestrationV2CheckpointFileSummary.Type;
+/** Sent only when the wire carries a prefix of `files`; the totals cover the full list. */
+export const OrchestrationV2CheckpointFileTotals = Schema.Struct({
+ count: NonNegativeInt,
+ additions: NonNegativeInt,
+ deletions: NonNegativeInt,
+});
+export type OrchestrationV2CheckpointFileTotals = typeof OrchestrationV2CheckpointFileTotals.Type;
+
export const OrchestrationV2Checkpoint = Schema.Struct({
id: CheckpointId,
threadId: ThreadId,
@@ -1361,6 +1369,7 @@ export const OrchestrationV2Checkpoint = Schema.Struct({
ref: CheckpointRef,
status: Schema.Literals(["ready", "missing", "error", "stale"]),
files: Schema.Array(OrchestrationV2CheckpointFileSummary),
+ filesTotal: Schema.optionalKey(OrchestrationV2CheckpointFileTotals),
capturedAt: Schema.DateTimeUtc,
});
export type OrchestrationV2Checkpoint = typeof OrchestrationV2Checkpoint.Type;
@@ -1659,6 +1668,7 @@ export const OrchestrationV2TurnItem = Schema.Union([
checkpointId: CheckpointId,
scopeId: CheckpointScopeId,
files: Schema.Array(OrchestrationV2CheckpointFileSummary),
+ filesTotal: Schema.optionalKey(OrchestrationV2CheckpointFileTotals),
/** Same as the files of checkpoint `checkpointId`. */
filesOmitted: DedupedBodyMarker,
}),
diff --git a/packages/shared/src/AgentScope.test.ts b/packages/shared/src/AgentScope.test.ts
index cf3ca64acce0a..2d9aca5ac1129 100644
--- a/packages/shared/src/AgentScope.test.ts
+++ b/packages/shared/src/AgentScope.test.ts
@@ -1,9 +1,12 @@
// @effect-diagnostics nodeBuiltinImport:off
import * as NodeChildProcess from "node:child_process";
+import * as NodeFS from "node:fs";
+import * as NodeOS from "node:os";
+import * as NodePath from "node:path";
import { describe, expect, it } from "@effect/vitest";
-import { AGENT_OOM_SCORE_ADJ, agentScopeCommand } from "./AgentScope.ts";
+import { AGENT_OOM_SCORE_ADJ, agentScopeCommand, agentScopeLauncherScript } from "./AgentScope.ts";
import * as HostProcess from "./HostProcess.ts";
describe.skipIf(HostProcess.Platform.defaultValue() !== "linux")("agentScopeCommand", () => {
@@ -25,5 +28,20 @@ describe.skipIf(HostProcess.Platform.defaultValue() !== "linux")("agentScopeComm
expect(Number(pid)).toBe(child.pid);
expect(Number(score)).toBe(AGENT_OOM_SCORE_ADJ);
expect(args).toEqual(["a b", "$HOME"]);
+
+ const directory = NodeFS.mkdtempSync(NodePath.join(NodeOS.tmpdir(), "t3-agent-launcher-"));
+ try {
+ const launcher = NodePath.join(directory, "launch's");
+ NodeFS.writeFileSync(launcher, agentScopeLauncherScript(launch), { mode: 0o700 });
+ const launched = NodeChildProcess.spawnSync(launcher, ["--flag", "it's"], {
+ encoding: "utf8",
+ });
+ const [launchedPid, launchedScore, ...launchedArgs] = launched.stdout.trimEnd().split("\n");
+ expect(Number(launchedPid)).toBe(launched.pid);
+ expect(Number(launchedScore)).toBe(AGENT_OOM_SCORE_ADJ);
+ expect(launchedArgs).toEqual(["a b", "$HOME", "--flag", "it's"]);
+ } finally {
+ NodeFS.rmSync(directory, { recursive: true, force: true });
+ }
});
});
diff --git a/packages/shared/src/AgentScope.ts b/packages/shared/src/AgentScope.ts
index 361bfc49108c6..ce2d58c017b25 100644
--- a/packages/shared/src/AgentScope.ts
+++ b/packages/shared/src/AgentScope.ts
@@ -74,6 +74,14 @@ export interface AgentSystemdScope {
readonly options: ReadonlyArray;
}
+const quoteShell = (value: string) => `'${value.replaceAll("'", `'\\''`)}'`;
+
+/** An executable file for launchers that take a path, not a command line, such as Playwright. */
+export function agentScopeLauncherScript(launch: AgentScopeCommand): string {
+ const command = [launch.command, ...launch.args].map(quoteShell).join(" ");
+ return `#!/bin/sh\nexec ${command} "$@"\n`;
+}
+
/** Exec at every step preserves the agent PID, stdio, and signals; systemd-run places it and descendants in the scope. */
export function agentScopeCommand(input: {
readonly command: string;
diff --git a/packages/shared/src/authScopeOptions.ts b/packages/shared/src/authScopeOptions.ts
index d443e50326d28..9a2261a09f3e9 100644
--- a/packages/shared/src/authScopeOptions.ts
+++ b/packages/shared/src/authScopeOptions.ts
@@ -104,3 +104,19 @@ export const AUTH_SCOPE_OPTIONS: ReadonlyArray<{
description: "Change managed tunnel connectivity.",
},
];
+
+/** Shown instead of the granted scopes while they are unknown, so every surface says the same thing. */
+export function sessionPermissionsUnavailableMessage(input: {
+ readonly connected: boolean;
+ /** The check errored; the client retries it while connected. */
+ readonly failed: boolean;
+ /** The environment answered that this session is not signed in; retrying will not change that. */
+ readonly rejected: boolean;
+}): string {
+ if (!input.connected) return "Connect to this environment to view this session’s permissions.";
+ if (input.rejected)
+ return "This environment did not accept your session. Reconnect to try again.";
+ return input.failed
+ ? "Could not verify your permissions. Retrying automatically."
+ : "Checking permissions…";
+}
diff --git a/packages/shared/src/composerContextReferences.ts b/packages/shared/src/composerContextReferences.ts
index e1b25bc094a66..88361606a9a64 100644
--- a/packages/shared/src/composerContextReferences.ts
+++ b/packages/shared/src/composerContextReferences.ts
@@ -132,6 +132,11 @@ export function collectComposerContextReferences(
export const ATTACHMENT_ONLY_BOOTSTRAP_PROMPT =
"[User attached one or more files without additional text. Respond using the conversation context and the attached files.]";
+/** Claude's ultrathink effort sends the prompt as `Ultrathink:\n`. */
+export function isAttachmentOnlyBootstrapPrompt(text: string): boolean {
+ return text.replace(/^Ultrathink:\s*/i, "") === ATTACHMENT_ONLY_BOOTSTRAP_PROMPT;
+}
+
export function isImageReferencesOnly(text: string): boolean {
const references = collectComposerContextReferences(text);
return (
diff --git a/packages/shared/src/nodeSqliteClient.test.ts b/packages/shared/src/nodeSqliteClient.test.ts
index 99eb8b37a215b..ac786993fa0bf 100644
--- a/packages/shared/src/nodeSqliteClient.test.ts
+++ b/packages/shared/src/nodeSqliteClient.test.ts
@@ -238,6 +238,27 @@ it.effect("lets the writer commit while a read-only transaction is open", () =>
),
);
+it.effect("serves another read while a read-only transaction is open", () =>
+ withReaderClient((sql) =>
+ Effect.gen(function* () {
+ const opened = yield* Deferred.make();
+ const other = yield* Effect.forkChild(
+ Deferred.await(opened).pipe(Effect.andThen(countEntries(sql).pipe(SqliteClient.readOnly))),
+ );
+ yield* sql
+ .withTransaction(
+ Effect.gen(function* () {
+ yield* countEntries(sql);
+ yield* Deferred.succeed(opened, undefined);
+ // One shared reader would make this join wait on the transaction forever.
+ assert.equal(yield* Fiber.join(other), 1);
+ }),
+ )
+ .pipe(SqliteClient.readOnly);
+ }),
+ ),
+);
+
// regression: #19226 read-only transactions run on the separate reader connection
it.effect("rejects a write in a read-only transaction", () =>
withReaderClient((sql) =>
@@ -333,6 +354,25 @@ it.effect("uses the writer when the reader cannot open the database", () =>
assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1);
assert.equal(yield* sql.withTransaction(countEntries(sql)).pipe(SqliteClient.readOnly), 1);
+
+ // A read beside an open read transaction must not start a worker on the next slot.
+ const held = yield* Deferred.make();
+ const release = yield* Deferred.make();
+ const holder = yield* sql
+ .withTransaction(
+ countEntries(sql).pipe(
+ Effect.andThen(Deferred.succeed(held, undefined)),
+ Effect.andThen(Deferred.await(release)),
+ ),
+ )
+ .pipe(SqliteClient.readOnly, Effect.forkChild);
+ yield* Deferred.await(held);
+ const other = yield* countEntries(sql).pipe(SqliteClient.readOnly, (effect) =>
+ Effect.forkChild(effect, { startImmediately: true }),
+ );
+ yield* Deferred.succeed(release, undefined);
+ yield* Fiber.join(holder);
+ assert.equal(yield* Fiber.join(other), 1);
assert.equal(startedWorkers.length, startedBefore + 1);
}),
),
diff --git a/packages/shared/src/nodeSqliteClient.ts b/packages/shared/src/nodeSqliteClient.ts
index a42943c065ac0..c0edfdb868d1f 100644
--- a/packages/shared/src/nodeSqliteClient.ts
+++ b/packages/shared/src/nodeSqliteClient.ts
@@ -293,8 +293,14 @@ const failed = (cause: unknown, message: string, operation: string) =>
// Read-transaction cleanup succeeds after worker exit because it can no longer change anything.
const endsReadTransaction = (sql: string) => /^\s*ROLLBACK\b|^\s*COMMIT\b/i.test(sql);
+// WAL readers run concurrently; one worker serialized every read behind the slowest transaction.
+const READER_POOL_SIZE = 4;
+
// Bind transactions to one worker; fail after its exit instead of resuming outside the transaction on a replacement.
-const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) {
+const makeReader = Effect.fnUntraced(function* (
+ options: SqliteClientConfig,
+ pool: { unavailable: boolean },
+) {
let current:
| {
readonly worker: NodeWorkerThreads.Worker;
@@ -302,7 +308,6 @@ const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) {
readonly started: Promise;
}
| undefined;
- let unavailable = false;
let nextRequestId = 0;
const start = () => {
@@ -439,8 +444,8 @@ const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) {
void reader.worker.terminate();
}
// @effect-diagnostics-next-line effectSucceedWithVoid:off - The connection result distinguishes undefined from void.
- if (unavailable) return Effect.succeed(undefined);
- unavailable = true;
+ if (pool.unavailable) return Effect.succeed(undefined);
+ pool.unavailable = true;
return Effect.logWarning("SQLite reader worker unavailable; reads use the main connection", {
reason,
}).pipe(Effect.as(undefined));
@@ -448,7 +453,7 @@ const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) {
return Effect.suspend((): Effect.Effect => {
// @effect-diagnostics-next-line effectSucceedWithVoid:off - The connection result distinguishes undefined from void.
- if (unavailable) return Effect.succeed(undefined);
+ if (pool.unavailable) return Effect.succeed(undefined);
let reader: NonNullable;
try {
reader = current ?? start();
@@ -467,7 +472,7 @@ const makeReader = Effect.fnUntraced(function* (options: SqliteClientConfig) {
}),
(startup): Effect.Effect =>
// A late ready event must not hand out a worker retired by another acquire timeout.
- unavailable || current !== reader
+ pool.unavailable || current !== reader
? // @effect-diagnostics-next-line effectSucceedWithVoid:off - The connection result distinguishes undefined from void.
Effect.succeed(undefined)
: Option.isSome(startup) && startup.value.type === "ready"
@@ -1073,21 +1078,48 @@ const make = Effect.fn("makeWithDatabase")(function* (
const semaphore = yield* Semaphore.make(1);
const connection = yield* makeConnection;
- const acquireReader =
- options.readerWorker && isFileDatabase ? yield* makeReader(options) : undefined;
- const readerSemaphore = yield* Semaphore.make(1);
+ // A reader that cannot start turns off the whole pool, so routing never picks a dead slot.
+ const readerPool = { unavailable: false };
+ const readers =
+ options.readerWorker && isFileDatabase
+ ? yield* Effect.forEach(Array.from({ length: READER_POOL_SIZE }), () =>
+ Effect.map(
+ Effect.all([makeReader(options, readerPool), Semaphore.make(1)]),
+ ([acquire, lock]) => ({
+ acquire,
+ lock,
+ claims: 0,
+ }),
+ ),
+ )
+ : [];
const writer = { name: "writer" as const, lock: semaphore, connection };
let activeWriter: ConnectionOwner | undefined;
- const route = Effect.withFiber((fiber) =>
- acquireReader !== undefined && Context.get(fiber.context, ReadOnly)
- ? Effect.map(acquireReader, (readerConnection) =>
+ // Callers must run route uninterruptibly and pass `claimed` a finalizer that runs `release` once.
+ const route = (claimed: (release: Effect.Effect) => Effect.Effect) =>
+ Effect.withFiber((fiber) => {
+ if (readers.length === 0 || readerPool.unavailable || !Context.get(fiber.context, ReadOnly)) {
+ return Effect.succeed(writer);
+ }
+ // Least claimed wins and ties go to the lowest slot, so idle slots never start a worker.
+ let slot = readers[0]!;
+ for (const candidate of readers) if (candidate.claims < slot.claims) slot = candidate;
+ slot.claims++;
+ const reader = slot;
+ return Effect.andThen(
+ claimed(
+ Effect.sync(() => {
+ reader.claims--;
+ }),
+ ),
+ Effect.map(reader.acquire, (readerConnection) =>
readerConnection === undefined
? writer
- : { name: "reader" as const, lock: readerSemaphore, connection: readerConnection },
- )
- : Effect.succeed(writer),
- );
+ : { name: "reader" as const, lock: reader.lock, connection: readerConnection },
+ ),
+ );
+ });
const acquired = (
target: { readonly name: Acquired["connection"] },
@@ -1101,88 +1133,110 @@ const make = Effect.fn("makeWithDatabase")(function* (
const writerBorrowBatchSize = 16;
let writerBorrowsSinceYield = 0;
const borrower: Borrower = (use) =>
- Effect.flatMap(route, (target) =>
- Effect.withFiber((fiber) => {
- const clock = fiber.getRef(Clock.Clock);
- const start = clock.monotonicTimeNanosUnsafe();
- const ownerAtWaitStart = target.name === "writer" ? activeWriter : undefined;
- const owner = target.name === "writer" ? connectionOwner(fiber) : undefined;
- const useConnection = Effect.suspend(() => {
- if (target.name === "writer") activeWriter = owner;
- if (
- fiber.getRef(References.TracerEnabled) &&
- fiber.getRef(Statement.SpanPropagationEnabled)
- ) {
- annotateAcquired(fiber, acquired(target, clock, start), ownerAtWaitStart);
- }
- return use(target.connection);
- });
- const useOwnedConnection =
- target.name === "writer"
- ? Effect.onExit(useConnection, () =>
- Effect.sync(() => {
- if (activeWriter === owner) activeWriter = undefined;
- }),
- )
- : useConnection;
- if (!options.checkpointWorker || target.name !== "writer") {
- return target.lock.withPermits(1)(useOwnedConnection);
- }
- const writerUse = Effect.onExit(useOwnedConnection, (exit) =>
- Effect.sync(() => {
- writerBorrowsSinceYield = Exit.hasInterrupts(exit)
- ? 0
- : Math.min(writerBorrowsSinceYield + 1, writerBorrowBatchSize);
- }),
- );
- return Effect.flatMap(target.lock.withPermitsIfAvailable(1)(writerUse), (result) =>
- Option.isSome(result)
- ? Effect.succeed(result.value)
- : target.lock.withPermits(1)(
- Effect.andThen(
- // Failed SQL also consumes the batch; cancellation hands off immediately.
- Effect.suspend(() => {
- if (writerBorrowsSinceYield < writerBorrowBatchSize) return Effect.void;
- writerBorrowsSinceYield = 0;
- return Effect.callback((resume) => {
- const immediate = setImmediate(() => resume(Effect.void));
- return Effect.sync(() => clearImmediate(immediate));
- });
- }),
- writerUse,
- ),
- ),
- );
- }),
- );
- const transactionAcquirer = Effect.flatMap(route, (target) =>
Effect.uninterruptibleMask((restore) => {
- const fiber = Fiber.getCurrent()!;
- const scope = Context.getUnsafe(fiber.context, Scope.Scope);
+ let release: Effect.Effect = Effect.void;
+ const claimed = (next: Effect.Effect) =>
+ Effect.sync(() => {
+ release = next;
+ });
+ return Effect.flatMap(route(claimed), (target) => restore(borrowTarget(target, use))).pipe(
+ Effect.ensuring(Effect.suspend(() => release)),
+ );
+ });
+ const borrowTarget = (
+ target: {
+ readonly name: Acquired["connection"];
+ readonly lock: Semaphore.Semaphore;
+ readonly connection: Connection;
+ },
+ use: (connection: Connection) => Effect.Effect,
+ ): Effect.Effect =>
+ Effect.withFiber((fiber) => {
const clock = fiber.getRef(Clock.Clock);
const start = clock.monotonicTimeNanosUnsafe();
- const pending = fiber.getRef(TransactionAcquire);
const ownerAtWaitStart = target.name === "writer" ? activeWriter : undefined;
const owner = target.name === "writer" ? connectionOwner(fiber) : undefined;
- const acquire = Effect.tap((pending?.restore ?? restore)(target.lock.take(1)), () => {
+ const useConnection = Effect.suspend(() => {
if (target.name === "writer") activeWriter = owner;
- if (pending !== undefined) {
- pending.acquired = acquired(target, clock, start);
- pending.ownerAtWaitStart = ownerAtWaitStart;
+ if (
+ fiber.getRef(References.TracerEnabled) &&
+ fiber.getRef(Statement.SpanPropagationEnabled)
+ ) {
+ annotateAcquired(fiber, acquired(target, clock, start), ownerAtWaitStart);
}
- return Scope.addFinalizer(
- scope,
- Effect.sync(() => {
- if (target.name === "writer" && activeWriter === owner) activeWriter = undefined;
- }).pipe(Effect.andThen(target.lock.release(1))),
- );
+ return use(target.connection);
});
- return Effect.as(
- // A scheduler yield after taking the permit must not precede its release finalizer.
- Effect.provideService(acquire, Scheduler.PreventSchedulerYield, true),
- target.connection,
+ const useOwnedConnection =
+ target.name === "writer"
+ ? Effect.onExit(useConnection, () =>
+ Effect.sync(() => {
+ if (activeWriter === owner) activeWriter = undefined;
+ }),
+ )
+ : useConnection;
+ if (!options.checkpointWorker || target.name !== "writer") {
+ return target.lock.withPermits(1)(useOwnedConnection);
+ }
+ const writerUse = Effect.onExit(useOwnedConnection, (exit) =>
+ Effect.sync(() => {
+ writerBorrowsSinceYield = Exit.hasInterrupts(exit)
+ ? 0
+ : Math.min(writerBorrowsSinceYield + 1, writerBorrowBatchSize);
+ }),
);
- }),
+ return Effect.flatMap(target.lock.withPermitsIfAvailable(1)(writerUse), (result) =>
+ Option.isSome(result)
+ ? Effect.succeed(result.value)
+ : target.lock.withPermits(1)(
+ Effect.andThen(
+ // Failed SQL also consumes the batch; cancellation hands off immediately.
+ Effect.suspend(() => {
+ if (writerBorrowsSinceYield < writerBorrowBatchSize) return Effect.void;
+ writerBorrowsSinceYield = 0;
+ return Effect.callback((resume) => {
+ const immediate = setImmediate(() => resume(Effect.void));
+ return Effect.sync(() => clearImmediate(immediate));
+ });
+ }),
+ writerUse,
+ ),
+ ),
+ );
+ });
+ const claimedUntilScopeCloses = (release: Effect.Effect) =>
+ Effect.withFiber((fiber) =>
+ Scope.addFinalizer(Context.getUnsafe(fiber.context, Scope.Scope), release),
+ );
+ const transactionAcquirer = Effect.uninterruptible(route(claimedUntilScopeCloses)).pipe(
+ Effect.flatMap((target) =>
+ Effect.uninterruptibleMask((restore) => {
+ const fiber = Fiber.getCurrent()!;
+ const scope = Context.getUnsafe(fiber.context, Scope.Scope);
+ const clock = fiber.getRef(Clock.Clock);
+ const start = clock.monotonicTimeNanosUnsafe();
+ const pending = fiber.getRef(TransactionAcquire);
+ const ownerAtWaitStart = target.name === "writer" ? activeWriter : undefined;
+ const owner = target.name === "writer" ? connectionOwner(fiber) : undefined;
+ const acquire = Effect.tap((pending?.restore ?? restore)(target.lock.take(1)), () => {
+ if (target.name === "writer") activeWriter = owner;
+ if (pending !== undefined) {
+ pending.acquired = acquired(target, clock, start);
+ pending.ownerAtWaitStart = ownerAtWaitStart;
+ }
+ return Scope.addFinalizer(
+ scope,
+ Effect.sync(() => {
+ if (target.name === "writer" && activeWriter === owner) activeWriter = undefined;
+ }).pipe(Effect.andThen(target.lock.release(1))),
+ );
+ });
+ return Effect.as(
+ // A scheduler yield after taking the permit must not precede its release finalizer.
+ Effect.provideService(acquire, Scheduler.PreventSchedulerYield, true),
+ target.connection,
+ );
+ }),
+ ),
);
const client = yield* Client.make({
diff --git a/packages/shared/src/threadFindText.test.ts b/packages/shared/src/threadFindText.test.ts
index 0dd6a87ae1dea..77a0874c21336 100644
--- a/packages/shared/src/threadFindText.test.ts
+++ b/packages/shared/src/threadFindText.test.ts
@@ -1,6 +1,7 @@
import { EnvironmentId, MessageId, ThreadId } from "@t3tools/contracts";
import { expect, it } from "vite-plus/test";
import { serializeAssistantCitation } from "./assistantCitations.ts";
+import { ATTACHMENT_ONLY_BOOTSTRAP_PROMPT } from "./composerContextReferences.ts";
import { searchableMessageSegments } from "./threadFindText.ts";
it("excludes review attachments rendered as context chips", () => {
@@ -31,6 +32,16 @@ it("excludes structured context chips without joining text across them", () => {
).toEqual(["before", "after"]);
});
+it("excludes the hidden prompt of an attachment-only message", () => {
+ expect(
+ searchableMessageSegments({
+ role: "user",
+ streaming: false,
+ text: ATTACHMENT_ONLY_BOOTSTRAP_PROMPT,
+ }),
+ ).toEqual([]);
+});
+
it("does not upgrade literal legacy tags in messages with structured context", () => {
const text =
"\n- Terminal 1 line 12:\n 12 | visible output\n";
diff --git a/packages/shared/src/threadFindText.ts b/packages/shared/src/threadFindText.ts
index a2e5b4dbb982d..3afec6e04729e 100644
--- a/packages/shared/src/threadFindText.ts
+++ b/packages/shared/src/threadFindText.ts
@@ -21,7 +21,10 @@ import {
resolveMarkdownFileLinkTarget,
} from "./markdownLinks.ts";
import { upgradeLegacyContextMessage } from "./composerContextLegacy.ts";
-import { parseComposerContextHref } from "./composerContextReferences.ts";
+import {
+ isAttachmentOnlyBootstrapPrompt,
+ parseComposerContextHref,
+} from "./composerContextReferences.ts";
import { assistantCitationLabel, parseAssistantCitationHref } from "./assistantCitations.ts";
import type { OrchestrationV2ConversationMessage } from "@t3tools/contracts";
import { proposedPlanTitle, stripDisplayedPlanMarkdown } from "./proposedPlanText.ts";
@@ -216,6 +219,8 @@ export function searchableMessageSegments(
skills: readonly InlineSkill[] = [],
): readonly string[] | null {
if (message.role === "user") {
+ // Clients hide this prompt, so find must not match text nobody can see.
+ if (isAttachmentOnlyBootstrapPrompt(message.text)) return [];
const text = message.context ? message.text : upgradeLegacyContextMessage(message.text).text;
return markdownThreadFindText(text, true, cwd, skills);
}
diff --git a/packages/source-control-bitbucket/src/server/BitbucketApi.test.ts b/packages/source-control-bitbucket/src/server/BitbucketApi.test.ts
index 26891c3548974..c0464276abf24 100644
--- a/packages/source-control-bitbucket/src/server/BitbucketApi.test.ts
+++ b/packages/source-control-bitbucket/src/server/BitbucketApi.test.ts
@@ -513,7 +513,7 @@ it.effect("never puts a saved token that is unsafe for an HTTP header on the wir
});
it.effect("reports saved credentials as configured when Bitbucket cannot confirm them", () => {
- const { layer } = makeLayer({
+ const { execute, layer } = makeLayer({
response: () => new Response(null, { status: 401 }),
env: { T3CODE_BITBUCKET_API_BASE_URL: "https://api.test.local/2.0" },
});
@@ -523,6 +523,7 @@ it.effect("reports saved credentials as configured when Bitbucket cannot confirm
const settings = yield* TestSourceControlHost.TestSourceControlHostSettings;
assert.strictEqual((yield* bitbucket.probeAuth).status, "unauthenticated");
+ assert.strictEqual(execute.mock.calls.length, 0);
yield* settings.update({
sourceControlHosts: { bitbucket: { accessToken: "saved-access-token" } },
diff --git a/packages/source-control-bitbucket/src/server/BitbucketApi.ts b/packages/source-control-bitbucket/src/server/BitbucketApi.ts
index 72077040aa870..863b206d95578 100644
--- a/packages/source-control-bitbucket/src/server/BitbucketApi.ts
+++ b/packages/source-control-bitbucket/src/server/BitbucketApi.ts
@@ -986,18 +986,25 @@ export const make = Effect.gen(function* () {
return BitbucketApi.of({
request,
- probeAuth: executeJson(
- "probeAuth",
- HttpClientRequest.get(apiUrl("/user")),
- BitbucketUserSchema,
- ).pipe(
- Effect.map((user) => ({
- status: "authenticated" as const,
- account: nonEmpty(user.username ?? user.display_name ?? user.account_id),
- host: Option.some("bitbucket.org"),
- detail: Option.none(),
- })),
- Effect.catch(() => currentCredential.pipe(Effect.map(authFromCredential))),
+ // Bitbucket answers anonymous /user with 401, so without a credential there is nothing to ask.
+ probeAuth: currentCredential.pipe(
+ Effect.flatMap((credential) =>
+ credential === null
+ ? Effect.succeed(authFromCredential(null))
+ : executeJson(
+ "probeAuth",
+ HttpClientRequest.get(apiUrl("/user")),
+ BitbucketUserSchema,
+ ).pipe(
+ Effect.map((user) => ({
+ status: "authenticated" as const,
+ account: nonEmpty(user.username ?? user.display_name ?? user.account_id),
+ host: Option.some("bitbucket.org"),
+ detail: Option.none(),
+ })),
+ Effect.catch(() => currentCredential.pipe(Effect.map(authFromCredential))),
+ ),
+ ),
),
listPullRequests: Effect.fn("BitbucketApi.listPullRequests")(function (input) {
return resolveRepository(input).pipe(
diff --git a/packages/source-control-core/src/server/discovery.ts b/packages/source-control-core/src/server/discovery.ts
index 66889ddc4f2ab..2f018bfa10ca3 100644
--- a/packages/source-control-core/src/server/discovery.ts
+++ b/packages/source-control-core/src/server/discovery.ts
@@ -75,6 +75,8 @@ type SourceControlCliRemoteRefinementSpec = SourceControlCliDiscoverySpec & {
};
const DEFAULT_PROBE_TIMEOUT_MS = 5_000;
+/** An API host that stalls must not hold the whole discovery answer behind it. */
+const API_PROBE_TIMEOUT = "3 seconds";
function probeTimeoutMs(spec: SourceControlCliDiscoverySpec): number {
return spec.probeTimeoutMs ?? DEFAULT_PROBE_TIMEOUT_MS;
@@ -236,6 +238,12 @@ export function probeSourceControlProvider(input: {
if (input.spec.type === "managed-cli") return input.spec.probe(input.cwd);
if (input.spec.type === "api") {
return input.spec.probeAuth.pipe(
+ Effect.timeoutOption(API_PROBE_TIMEOUT),
+ Effect.map(
+ Option.getOrElse(() =>
+ unknownAuth("The hosting service did not answer in time. Refresh to try again."),
+ ),
+ ),
Effect.map(
(auth) =>
({
@@ -253,20 +261,15 @@ export function probeSourceControlProvider(input: {
const spec = input.spec;
- return probeCli({
- spec,
- process: input.process,
- cwd: input.cwd,
- }).pipe(
- Effect.flatMap((item) => {
- if (item.status !== "available") {
- return Effect.succeed({
- ...item,
- auth: unknownAuth("Hosting integration command was not found on the server PATH."),
- } satisfies SourceControlProviderDiscoveryItem);
- }
-
- return input.process
+ // The auth check does not depend on the version output, so both spawn at once.
+ return Effect.all(
+ [
+ probeCli({
+ spec,
+ process: input.process,
+ cwd: input.cwd,
+ }),
+ input.process
.run({
operation: "source-control.discovery.auth",
command: spec.executable,
@@ -278,21 +281,24 @@ export function probeSourceControlProvider(input: {
appendTruncationMarker: true,
})
.pipe(
- Effect.map(
- (result) =>
- ({
- ...item,
- auth: spec.parseAuth(result),
- }) satisfies SourceControlProviderDiscoveryItem,
- ),
+ Effect.map(spec.parseAuth),
Effect.catch((cause) =>
- Effect.succeed({
- ...item,
- auth: unknownAuth(Option.getOrUndefined(detailFromCause(cause))),
- } satisfies SourceControlProviderDiscoveryItem),
+ Effect.succeed(unknownAuth(Option.getOrUndefined(detailFromCause(cause)))),
),
- );
- }),
+ ),
+ ],
+ { concurrency: "unbounded" },
+ ).pipe(
+ Effect.map(
+ ([item, auth]) =>
+ ({
+ ...item,
+ auth:
+ item.status === "available"
+ ? auth
+ : unknownAuth("Hosting integration command was not found on the server PATH."),
+ }) satisfies SourceControlProviderDiscoveryItem,
+ ),
);
}
diff --git a/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.test.ts b/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.test.ts
index 7c257f037003e..113a388e853bb 100644
--- a/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.test.ts
+++ b/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.test.ts
@@ -1226,43 +1226,70 @@ describe("GitCafePullRequestProvider", () => {
it.effect("recounts summary lines only when the compared commits move", () => {
const landedBase = "fedcba9876543210fedcba9876543210fedcba98";
+ const mainTip = "0123012301230123012301230123012301230123";
let comparisonBaseOid = baseOid;
+ let tips = { targetBranch: "main", observedBaseOid: baseOid };
let pullReads = 0;
+ let listingCut = true;
+ let hunksDown = false;
+ const hunks = (request: SentRequest) => ({
+ version: 4,
+ headOid,
+ comparisonBaseOid,
+ items: (request.body as { paths: ReadonlyArray }).paths.map((path) => ({
+ path,
+ additions: comparisonBaseOid === baseOid ? 3 : 1,
+ deletions: 1,
+ })),
+ });
const server = fakeGitCafe({
// A comment moves the version without moving the compared commits.
- "GET /repos/owner/repo/pulls/7": () => ({ ...pull, version: 4 + pullReads++ }),
- "GET /repos/owner/repo/pulls/7/changes": () => ({
- version: 4,
- headOid,
- comparisonBaseOid,
- items: [{ path: "a.ts" }, { path: "b.ts" }],
- next: null,
- }),
- "POST /repos/owner/repo/pulls/7/diff-files": (request: SentRequest) => ({
+ "GET /repos/owner/repo/pulls/7": () => ({ ...pull, ...tips, version: 4 + pullReads++ }),
+ "GET /repos/owner/repo/pulls/7/changes": (request: SentRequest) => ({
version: 4,
headOid,
comparisonBaseOid,
- items: (request.body as { paths: ReadonlyArray }).paths.map((path) => ({
- path,
- additions: comparisonBaseOid === baseOid ? 3 : 1,
- deletions: 1,
- })),
+ ...(!listingCut
+ ? { items: [{ path: "a.ts" }, { path: "b.ts" }], next: null }
+ : request.query.has("after")
+ ? { items: [], next: "more" }
+ : { items: [{ path: "a.ts" }], next: "more" }),
}),
+ get "POST /repos/owner/repo/pulls/7/diff-files"() {
+ return hunksDown
+ ? new Response(encodeJson({ type: "https://cafe.sh/errors/conflict" }), { status: 409 })
+ : hunks;
+ },
});
- const hunkReads = () =>
- server.sent.filter((request) => request.path.endsWith("/diff-files")).length;
+ const reads = (suffix: string) =>
+ server.sent.filter((request) => request.path.endsWith(suffix)).length;
return Effect.gen(function* () {
const provider = yield* GitCafePullRequestProvider.make;
+ // A listing that stops before its last page holds nothing for its commits.
+ yield* provider.getChangeRequestSummary!(target);
+ listingCut = false;
+ hunksDown = true;
+ // Unread hunks leave the lines unknown, and are read again next time.
+ const unread = yield* provider.getChangeRequestSummary!(target);
+ assert.strictEqual(unread.additions, undefined);
+ hunksDown = false;
const first = yield* provider.getChangeRequestSummary!(target);
const second = yield* provider.getChangeRequestSummary!(target);
assert.deepInclude(first, { additions: 6, deletions: 2, changedFiles: 2 });
assert.deepInclude(second, { additions: 6, deletions: 2, changedFiles: 2 });
- assert.strictEqual(hunkReads(), 1);
- // A landed parent layer moves the comparison base under the same head.
- comparisonBaseOid = landedBase;
+ // Neither the head nor the target tip moved, so the file list is not read again.
+ assert.deepStrictEqual([reads("/changes"), reads("/diff-files")], [4, 3]);
+ // The target branch advancing keeps the merge base, so only the file list is read.
+ tips = { ...tips, observedBaseOid: mainTip };
const third = yield* provider.getChangeRequestSummary!(target);
- assert.deepInclude(third, { additions: 2, deletions: 2, changedFiles: 2 });
- assert.strictEqual(hunkReads(), 2);
+ assert.deepInclude(third, { additions: 6, deletions: 2, changedFiles: 2 });
+ assert.deepStrictEqual([reads("/changes"), reads("/diff-files")], [5, 3]);
+ // A landed parent layer retargets the pull and moves the comparison base under the same head.
+ comparisonBaseOid = landedBase;
+ tips = { targetBranch: "trunk", observedBaseOid: mainTip };
+ const fourth = yield* provider.getChangeRequestSummary!(target);
+ assert.deepInclude(fourth, { additions: 2, deletions: 2, changedFiles: 2 });
+ assert.deepStrictEqual([reads("/changes"), reads("/diff-files")], [6, 4]);
}).pipe(Effect.provide(server.layer));
});
diff --git a/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.ts b/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.ts
index 1d4b7338f3c12..e87a4e154d69b 100644
--- a/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.ts
+++ b/packages/source-control-gitcafe/src/server/GitCafePullRequestProvider.ts
@@ -636,11 +636,19 @@ export const make = Effect.gen(function* () {
// so hunk totals are held per compared pair of commits. The file list that names the pair is
// cheap; a retarget or a landed parent layer moves the base without moving the head.
const lineTotals = new Map();
+ // The compared pair follows from the head and the target tip the pull reports, so while neither
+ // moves the stats are reused without listing the files again.
+ const statsByTips = new Map<
+ string,
+ { readonly changedFiles: number; readonly additions?: number; readonly deletions?: number }
+ >();
+ const remember = (cache: Map, key: K, value: V) => {
+ if (cache.size >= LINE_STATS_CACHE_SIZE) cache.delete(cache.keys().next().value!);
+ cache.set(key, value);
+ };
- const readLineStats = Effect.fn("GitCafePullRequestProvider.readLineStats")(function* (
- input: PullRef,
- version: number,
- ) {
+ // readLineStats owns the span.
+ const countLines = Effect.fnUntraced(function* (input: PullRef, version: number) {
const path = yield* pullPath("getChangeRequest", input);
const files: Array = [];
let revision: { version: number; headOid: string; baseOid: string } | undefined;
@@ -667,13 +675,23 @@ export const make = Effect.gen(function* () {
after = batch.next;
if (after === null || after === previous || batch.items.length === 0) break;
}
- if (revision === undefined) return { changedFiles: 0, additions: 0, deletions: 0 };
+ if (revision === undefined)
+ return {
+ headOid: null,
+ complete: true,
+ stats: { changedFiles: 0, additions: 0, deletions: 0 },
+ };
const key = `${input.host} ${input.repository.toLowerCase()} ${input.number} ${revision.baseOid} ${revision.headOid}`;
const held = lineTotals.get(key);
- if (held !== undefined) return { changedFiles: files.length, ...held };
+ const { headOid } = revision;
+ // A listing that stopped before its last page counts only part of the diff, so nothing is held.
+ const listed = after === null;
+ if (held !== undefined)
+ return { headOid, complete: listed, stats: { changedFiles: files.length, ...held } };
+ const tooMany = files.length > LINE_STATS_MAX_FILES;
const details = files.some((file) => file.additions !== undefined)
? Option.some(files)
- : files.length > LINE_STATS_MAX_FILES
+ : tooMany
? Option.none()
: yield* Effect.forEach(
chunk(detailablePaths(files), DIFF_FILES_BATCH),
@@ -684,16 +702,33 @@ export const make = Effect.gen(function* () {
Effect.option,
);
// Unread hunks leave the lines unknown rather than zero; the file count still stands.
- if (Option.isNone(details)) return { changedFiles: files.length };
+ if (Option.isNone(details))
+ return { headOid, complete: listed && tooMany, stats: { changedFiles: files.length } };
const counted = details.value;
const totals = {
additions: counted.reduce((total, file) => total + (file.additions ?? 0), 0),
deletions: counted.reduce((total, file) => total + (file.deletions ?? 0), 0),
};
- if (lineTotals.size >= LINE_STATS_CACHE_SIZE)
- lineTotals.delete(lineTotals.keys().next().value!);
- lineTotals.set(key, totals);
- return { changedFiles: files.length, ...totals };
+ if (listed) remember(lineTotals, key, totals);
+ return { headOid, complete: listed, stats: { changedFiles: files.length, ...totals } };
+ });
+
+ const readLineStats = Effect.fn("GitCafePullRequestProvider.readLineStats")(function* (
+ input: PullRef,
+ pull: Json.GitCafePullDetail,
+ ) {
+ const tipsKey =
+ pull.headOid === null || pull.observedBaseOid == null
+ ? null
+ : `${input.host} ${input.repository.toLowerCase()} ${input.number} ${pull.targetBranch} ${pull.observedBaseOid} ${pull.headOid}`;
+ const heldStats = tipsKey === null ? undefined : statsByTips.get(tipsKey);
+ if (heldStats !== undefined) return heldStats;
+ const stats = yield* countLines(input, pull.version);
+ // A failed hunk read is retried next time rather than held as unknown, and a head that moved
+ // between the two reads would hold another pair's stats.
+ if (tipsKey !== null && stats.complete && stats.headOid === pull.headOid)
+ remember(statsByTips, tipsKey, stats.stats);
+ return stats.stats;
});
const merge = Effect.fn("GitCafePullRequestProvider.merge")(function* (
@@ -870,7 +905,7 @@ export const make = Effect.gen(function* () {
const changes =
pull.headOid === null
? null
- : yield* readLineStats(input, pull.version).pipe(Effect.orElseSucceed(() => null));
+ : yield* readLineStats(input, pull).pipe(Effect.orElseSucceed(() => null));
// A count that could not be read is absent rather than zero.
const {
additions: _additions,
@@ -904,7 +939,7 @@ export const make = Effect.gen(function* () {
),
pull.headOid === null
? Effect.succeed(null)
- : readLineStats(input, pull.version).pipe(Effect.orElseSucceed(() => null)),
+ : readLineStats(input, pull).pipe(Effect.orElseSucceed(() => null)),
// Reviewers need `pr:review:read`, which a contents-only token lacks.
readPages("getChangeRequest", input, `${path}/reviewers`, Json.GitCafeActorEntries).pipe(
// @effect-diagnostics-next-line catchConditionalRefailToCatchIf:off - catchIf widens this paginated result to any in tsc-rs.