feat: enhance OpenCodeAdapter with session management and cleanup logic

Author: ronak-guliani

apps/server/src/provider/Layers/OpenCodeAdapter.test.ts

@@ -480,7 +480,25 @@ it.layer(OpenCodeAdapterTestLayer)("OpenCodeAdapterLive", (it) => {
 
       assert.equal(sessions.length, 1);
       assert.equal(sessions[0]?.threadId, "thread-native-log-failure");
-      assert.deepEqual(runtimeMock.state.closeCalls, []);
+      assert.deepEqual(runtimeMock.state.closeCalls, ["http://127.0.0.1:9999"]);
     }),
   );
 });
+
+it.effect("OpenCodeAdapterLive stops active sessions when the adapter layer is released", () =>
+  Effect.gen(function* () {
+    yield* Effect.scoped(
+      Effect.gen(function* () {
+        const adapter = yield* OpenCodeAdapter;
+        yield* adapter.startSession({
+          provider: "opencode",
+          threadId: asThreadId("thread-finalizer"),
+          runtimeMode: "full-access",
+        });
+      }).pipe(Effect.provide(OpenCodeAdapterTestLayer)),
+    );
+
+    assert.deepEqual(runtimeMock.state.abortCalls, ["http://127.0.0.1:9999/session"]);
+    assert.deepEqual(runtimeMock.state.closeCalls, ["http://127.0.0.1:9999"]);
+  }),
+);

apps/server/src/provider/Layers/OpenCodeAdapter.ts

@@ -40,11 +40,7 @@ import {
 
 const PROVIDER = "opencode" as const;
 const OPEN_CODE_STALL_WARNING_MS = 15_000;
-
-interface OpenCodeTurnSnapshot {
-  readonly id: TurnId;
-  readonly items: Array<unknown>;
-}
+const EXTERNAL_OPENCODE_ABORT_GRACE_MS = 2_000;
 
 interface OpenCodeSessionContext {
   session: ProviderSession;
@@ -58,7 +54,6 @@ interface OpenCodeSessionContext {
   readonly partById: Map<string, Part>;
   readonly emittedTextByPartId: Map<string, string>;
   readonly completedAssistantPartIds: Set<string>;
-  readonly turns: Array<OpenCodeTurnSnapshot>;
   activeTurnId: TurnId | undefined;
   activeAgent: string | undefined;
   activeVariant: string | undefined;
@@ -177,29 +172,22 @@ function mapPermissionDecision(reply: "once" | "always" | "reject"): string {
   }
 }
 
-function resolveTurnSnapshot(
-  context: OpenCodeSessionContext,
-  turnId: TurnId,
-): OpenCodeTurnSnapshot {
-  const existing = context.turns.find((turn) => turn.id === turnId);
-  if (existing) {
-    return existing;
-  }
-
-  const created: OpenCodeTurnSnapshot = { id: turnId, items: [] };
-  context.turns.push(created);
-  return created;
-}
-
-function appendTurnItem(
-  context: OpenCodeSessionContext,
-  turnId: TurnId | undefined,
-  item: unknown,
-): void {
-  if (!turnId) {
-    return;
-  }
-  resolveTurnSnapshot(context, turnId).items.push(item);
+/**
+ * Drop per-turn streaming bookkeeping after a turn ends. Long sessions otherwise
+ * accumulate full assistant/reasoning text and tool-call state in `partById` /
+ * `emittedTextByPartId` for every turn ever processed, which can grow into the
+ * multi-GB range for sustained use.
+ *
+ * We intentionally drop *all* parts: events that arrive after the turn ends for
+ * a previously-seen part are no-ops in the existing handlers (they guard on
+ * `partById.get(...)` returning `undefined`), and assistant text for a closed
+ * turn has already been emitted as `item.completed`.
+ */
+function pruneCompletedTurnState(context: OpenCodeSessionContext): void {
+  context.partById.clear();
+  context.emittedTextByPartId.clear();
+  context.completedAssistantPartIds.clear();
+  context.messageRoleById.clear();
 }
 
 function ensureSessionContext(
@@ -509,12 +497,20 @@ async function stopOpenCodeContext(context: OpenCodeSessionContext): Promise<voi
     context.stallWarningTimer = undefined;
   }
   context.eventsAbortController.abort();
-  try {
-    await context.client.session
-      .abort({ sessionID: context.openCodeSessionId })
-      .catch(() => undefined);
-  } catch {}
+  if (context.server.external) {
+    // External servers outlive us: try to abort the remote session politely,
+    // but do not block teardown if the RPC is wedged.
+    await Promise.race([
+      context.client.session.abort({ sessionID: context.openCodeSessionId }).catch(() => undefined),
+      new Promise<void>((resolve) => setTimeout(resolve, EXTERNAL_OPENCODE_ABORT_GRACE_MS)),
+    ]);
+    context.server.close();
+    return;
+  }
+  // Local server: skip the abort RPC entirely — we own the process and are
+  // about to terminate it, so the HTTP call would just race a dropped socket.
   context.server.close();
+  pruneCompletedTurnState(context);
 }
 
 export function makeOpenCodeAdapterLive(_options?: OpenCodeAdapterLiveOptions) {
@@ -531,6 +527,8 @@ export function makeOpenCodeAdapterLive(_options?: OpenCodeAdapterLiveOptions) {
               stream: "native",
             })
           : undefined);
+      const managedNativeEventLogger =
+        _options?.nativeEventLogger === undefined ? nativeEventLogger : undefined;
       const runtimeEvents = yield* Queue.unbounded<ProviderRuntimeEvent>();
       const sessions = new Map<ThreadId, OpenCodeSessionContext>();
 
@@ -593,7 +591,9 @@ export function makeOpenCodeAdapterLive(_options?: OpenCodeAdapterLiveOptions) {
         context.stopped = true;
         clearTurnStallWarning(context);
         sessions.delete(context.session.threadId);
+        context.eventsAbortController.abort();
         context.server.close();
+        pruneCompletedTurnState(context);
         void logWarning("opencode.session.unexpected-exit", {
           message,
           ...turnObservabilitySnapshot(context),
@@ -861,7 +861,6 @@ export function makeOpenCodeAdapterLive(_options?: OpenCodeAdapterLiveOptions) {
                             : "item.updated",
                       payload,
                     };
-                    appendTurnItem(context, turnId, part);
                     await emitPromise(runtimeEvent);
                   }
                   break;
@@ -1022,6 +1021,7 @@ export function makeOpenCodeAdapterLive(_options?: OpenCodeAdapterLiveOptions) {
                     clearTurnStallWarning(context);
                     context.activeTurnId = undefined;
                     context.activeTurnStartedAt = undefined;
+                    pruneCompletedTurnState(context);
                     updateProviderSession(
                       context,
                       { status: "ready" },
@@ -1044,6 +1044,7 @@ export function makeOpenCodeAdapterLive(_options?: OpenCodeAdapterLiveOptions) {
                   clearTurnStallWarning(context);
                   context.activeTurnId = undefined;
                   context.activeTurnStartedAt = undefined;
+                  pruneCompletedTurnState(context);
                   updateProviderSession(
                     context,
                     {
@@ -1238,7 +1239,6 @@ export function makeOpenCodeAdapterLive(_options?: OpenCodeAdapterLiveOptions) {
             emittedTextByPartId: new Map(),
             messageRoleById: new Map(),
             completedAssistantPartIds: new Set(),
-            turns: [],
             activeTurnId: undefined,
             activeAgent: undefined,
             activeVariant: undefined,
@@ -1397,6 +1397,7 @@ export function makeOpenCodeAdapterLive(_options?: OpenCodeAdapterLiveOptions) {
           context.activeAgent = undefined;
           context.activeVariant = undefined;
           context.activeTurnStartedAt = undefined;
+          pruneCompletedTurnState(context);
           updateProviderSession(
             context,
             {
@@ -1671,6 +1672,18 @@ export function makeOpenCodeAdapterLive(_options?: OpenCodeAdapterLiveOptions) {
             }),
         });
 
+      yield* Effect.addFinalizer(() =>
+        stopAll().pipe(
+          Effect.catchCause((cause) =>
+            Effect.logWarning("opencode.session.stop-all-finalizer-failed", {
+              cause: Cause.pretty(cause),
+            }),
+          ),
+          Effect.tap(() => Queue.shutdown(runtimeEvents)),
+          Effect.tap(() => managedNativeEventLogger?.close() ?? Effect.void),
+        ),
+      );
+
       return {
         provider: PROVIDER,
         capabilities: {

apps/server/src/provider/opencodeRuntime.ts

@@ -378,7 +378,16 @@ export async function startOpenCodeServerProcess(input: {
       return;
     }
     closed = true;
-    child.kill();
+    const forceKillTimer = setTimeout(() => {
+      if (child.exitCode === null && child.signalCode === null) {
+        child.kill("SIGKILL");
+      }
+    }, 1_000);
+    forceKillTimer.unref();
+    child.once("exit", () => clearTimeout(forceKillTimer));
+    child.stdout?.destroy();
+    child.stderr?.destroy();
+    child.kill("SIGTERM");
   };
 
   const url = await new Promise<string>((resolve, reject) => {
@@ -441,6 +450,12 @@ export async function startOpenCodeServerProcess(input: {
     child.once("close", onClose);
   });
 
+  // After startup, keep draining the child's stdio so its OS pipe buffers
+  // never fill (which would block the OpenCode server on its next write).
+  // We don't retain the data \u2014 only listeners that count as consumers.
+  child.stdout.on("data", () => {});
+  child.stderr.on("data", () => {});
+
   return {
     url,
     process: child,

package.json

@@ -53,6 +53,7 @@
     "dist:desktop:win": "node scripts/build-desktop-artifact.ts --platform win --target nsis",
     "dist:desktop:win:arm64": "node scripts/build-desktop-artifact.ts --platform win --target nsis --arch arm64",
     "dist:desktop:win:x64": "node scripts/build-desktop-artifact.ts --platform win --target nsis --arch x64",
+    "install:desktop:local": "bash scripts/install-desktop-local.sh",
     "release:smoke": "node scripts/release-smoke.ts",
     "clean": "rm -rf node_modules apps/*/node_modules packages/*/node_modules apps/*/dist apps/*/dist-electron packages/*/dist .turbo apps/*/.turbo packages/*/.turbo",
     "sync:vscode-icons": "node scripts/sync-vscode-icons.mjs"

scripts/install-desktop-local.sh

@@ -0,0 +1,116 @@
+#!/usr/bin/env bash
+#
+# Build the arm64 macOS DMG and install it into /Applications, replacing any
+# previous installation. Quits the running app first, clears quarantine, and
+# launches the freshly installed build.
+#
+# Usage:
+#   scripts/install-desktop-local.sh           # build + install + launch
+#   scripts/install-desktop-local.sh --no-build  # reuse the existing DMG
+#   scripts/install-desktop-local.sh --no-launch # skip the open at the end
+#
+set -euo pipefail
+
+REPO_ROOT="$(cd "$(dirname "${BASH_SOURCE[0]}")/.." && pwd)"
+APP_NAME="T3 Code (Alpha)"
+APP_BUNDLE="${APP_NAME}.app"
+INSTALL_DEST="/Applications/${APP_BUNDLE}"
+RELEASE_DIR="${REPO_ROOT}/release"
+
+DO_BUILD=1
+DO_LAUNCH=1
+for arg in "$@"; do
+  case "$arg" in
+    --no-build) DO_BUILD=0 ;;
+    --no-launch) DO_LAUNCH=0 ;;
+    -h|--help)
+      sed -n '2,12p' "$0"
+      exit 0
+      ;;
+    *)
+      echo "Unknown argument: $arg" >&2
+      exit 2
+      ;;
+  esac
+done
+
+if [[ "$(uname)" != "Darwin" ]]; then
+  echo "This script only runs on macOS." >&2
+  exit 1
+fi
+
+HOST_ARCH="$(uname -m)"
+if [[ "$HOST_ARCH" != "arm64" ]]; then
+  echo "Warning: host arch is ${HOST_ARCH}; this script builds an arm64 DMG." >&2
+fi
+
+log() { printf '\n[install-desktop] %s\n' "$*"; }
+
+log "Quitting any running ${APP_NAME} instance..."
+osascript -e "tell application \"${APP_NAME}\" to quit" >/dev/null 2>&1 || true
+# Give the app a moment to exit cleanly, then force-kill any stragglers.
+sleep 1
+pkill -f "${APP_BUNDLE}/Contents/MacOS/" >/dev/null 2>&1 || true
+
+if [[ "$DO_BUILD" -eq 1 ]]; then
+  log "Building arm64 DMG (this takes ~1 minute)..."
+  rm -f "${RELEASE_DIR}"/T3-Code-*-arm64.dmg \
+        "${RELEASE_DIR}"/T3-Code-*-arm64.dmg.blockmap \
+        "${RELEASE_DIR}"/T3-Code-*-arm64.zip \
+        "${RELEASE_DIR}"/T3-Code-*-arm64.zip.blockmap
+  ( cd "$REPO_ROOT" && bun run dist:desktop:dmg:arm64 )
+fi
+
+DMG_PATH="$(ls -t "${RELEASE_DIR}"/T3-Code-*-arm64.dmg 2>/dev/null | head -n 1 || true)"
+if [[ -z "$DMG_PATH" || ! -f "$DMG_PATH" ]]; then
+  echo "No arm64 DMG found in ${RELEASE_DIR}." >&2
+  echo "Re-run without --no-build to produce one." >&2
+  exit 1
+fi
+log "Using DMG: ${DMG_PATH}"
+
+# Mount the DMG into a temporary mount point and ensure we always detach it,
+# even if ditto/cp fails.
+MOUNT_POINT=""
+cleanup() {
+  if [[ -n "$MOUNT_POINT" && -d "$MOUNT_POINT" ]]; then
+    hdiutil detach "$MOUNT_POINT" -quiet || hdiutil detach "$MOUNT_POINT" -force -quiet || true
+  fi
+}
+trap cleanup EXIT
+
+log "Mounting DMG..."
+ATTACH_OUTPUT="$(hdiutil attach -nobrowse -readonly -plist "$DMG_PATH")"
+# Pull the first /Volumes/* mount point out of the plist.
+MOUNT_POINT="$(printf '%s' "$ATTACH_OUTPUT" \
+  | /usr/bin/awk '/<string>\/Volumes\//{ sub(/.*<string>/,""); sub(/<\/string>.*/,""); print; exit }')"
+if [[ -z "$MOUNT_POINT" || ! -d "$MOUNT_POINT" ]]; then
+  echo "Failed to determine DMG mount point." >&2
+  exit 1
+fi
+log "Mounted at: ${MOUNT_POINT}"
+
+SRC_APP="${MOUNT_POINT}/${APP_BUNDLE}"
+if [[ ! -d "$SRC_APP" ]]; then
+  echo "Source app not found at ${SRC_APP}." >&2
+  ls -la "$MOUNT_POINT" >&2
+  exit 1
+fi
+
+log "Replacing ${INSTALL_DEST}..."
+rm -rf "$INSTALL_DEST"
+ditto "$SRC_APP" "$INSTALL_DEST"
+
+log "Clearing quarantine attributes..."
+xattr -dr com.apple.quarantine "$INSTALL_DEST" 2>/dev/null || true
+
+INSTALLED_VERSION="$(/usr/libexec/PlistBuddy -c 'Print :CFBundleShortVersionString' \
+  "${INSTALL_DEST}/Contents/Info.plist" 2>/dev/null || echo 'unknown')"
+log "Installed ${APP_NAME} v${INSTALLED_VERSION}"
+
+if [[ "$DO_LAUNCH" -eq 1 ]]; then
+  log "Launching..."
+  open -a "$INSTALL_DEST"
+fi
+
+log "Done."