[Java] Add in-process FFI runtime host lifecycle and stream transport primitives (#2233)

* Initial plan

* java: add FFI runtime host streams and lifecycle

Co-authored-by: edburns <75821+edburns@users.noreply.github.com>

* fix: address Copilot review comments on PR #2233

1. FfiRuntimeHost: serialize start() with close() via operationLock to
   prevent race where close() completes while hostStart blocks, then
   start() publishes handles after teardown.

2. FfiRuntimeHostTest: add Javadoc explaining the native integration
   test is intentionally skipped in CI and how to run it locally.

3. JsonRpcClient.fromStreams(): take ownership of supplied streams and
   close them in close() to prevent resource leaks.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

---------

Co-authored-by: copilot-swe-agent[bot] <198982749+Copilot@users.noreply.github.com>
Co-authored-by: edburns <75821+edburns@users.noreply.github.com>
Co-authored-by: Ed Burns <edburns@microsoft.com>
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>

Author: 3 people

java/sdk/src/main/java/com/github/copilot/JsonRpcClient.java

@@ -46,17 +46,24 @@ class JsonRpcClient implements AutoCloseable {
     private final OutputStream outputStream;
     private final Socket socket;
     private final Process process;
+    private final boolean ownsStreams;
     private final AtomicLong requestIdCounter = new AtomicLong(0);
     private final Map<Long, CompletableFuture<JsonNode>> pendingRequests = new ConcurrentHashMap<>();
     private final Map<String, BiConsumer<String, JsonNode>> notificationHandlers = new ConcurrentHashMap<>();
     private final ExecutorService readerExecutor;
     private volatile boolean running = true;
 
     private JsonRpcClient(InputStream inputStream, OutputStream outputStream, Socket socket, Process process) {
+        this(inputStream, outputStream, socket, process, false);
+    }
+
+    private JsonRpcClient(InputStream inputStream, OutputStream outputStream, Socket socket, Process process,
+            boolean ownsStreams) {
         this.inputStream = inputStream;
         this.outputStream = outputStream;
         this.socket = socket;
         this.process = process;
+        this.ownsStreams = ownsStreams;
         this.readerExecutor = Executors.newSingleThreadExecutor(r -> {
             Thread t = new Thread(r, "jsonrpc-reader");
             t.setDaemon(true);
@@ -93,6 +100,15 @@ public static JsonRpcClient fromSocket(Socket socket) throws IOException {
         return new JsonRpcClient(socket.getInputStream(), socket.getOutputStream(), socket, null);
     }
 
+    /**
+     * Creates a JSON-RPC client over arbitrary input/output streams. The client
+     * takes ownership of the streams and closes them when {@link #close()} is
+     * called.
+     */
+    public static JsonRpcClient fromStreams(InputStream inputStream, OutputStream outputStream) {
+        return new JsonRpcClient(inputStream, outputStream, null, null, true);
+    }
+
     /**
      * Registers a handler for JSON-RPC method calls (requests/notifications from
      * server).
@@ -344,6 +360,19 @@ public void close() {
         if (process != null) {
             process.destroy();
         }
+
+        if (ownsStreams) {
+            try {
+                inputStream.close();
+            } catch (IOException e) {
+                LOG.log(Level.FINE, "Error closing input stream", e);
+            }
+            try {
+                outputStream.close();
+            } catch (IOException e) {
+                LOG.log(Level.FINE, "Error closing output stream", e);
+            }
+        }
     }
 
     public boolean isConnected() {

java/sdk/src/main/java/com/github/copilot/ffi/FfiOutputStream.java

@@ -0,0 +1,63 @@
+/*---------------------------------------------------------------------------------------------
+ *  Copyright (c) Microsoft Corporation. All rights reserved.
+ *--------------------------------------------------------------------------------------------*/
+
+package com.github.copilot.ffi;
+
+import java.io.IOException;
+import java.io.OutputStream;
+import java.util.Arrays;
+import java.util.Objects;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
+import java.util.concurrent.locks.ReentrantLock;
+
+final class FfiOutputStream extends OutputStream {
+
+    private final NativeBinding nativeBinding;
+    private final AtomicInteger connectionId;
+    private final AtomicBoolean closing;
+    private final ReentrantLock operationLock;
+
+    FfiOutputStream(NativeBinding nativeBinding, AtomicInteger connectionId, AtomicBoolean closing,
+            ReentrantLock operationLock) {
+        this.nativeBinding = Objects.requireNonNull(nativeBinding, "nativeBinding must not be null");
+        this.connectionId = Objects.requireNonNull(connectionId, "connectionId must not be null");
+        this.closing = Objects.requireNonNull(closing, "closing must not be null");
+        this.operationLock = Objects.requireNonNull(operationLock, "operationLock must not be null");
+    }
+
+    @Override
+    public void write(int b) throws IOException {
+        write(new byte[]{(byte) b}, 0, 1);
+    }
+
+    @Override
+    public void write(byte[] b, int off, int len) throws IOException {
+        Objects.requireNonNull(b, "buffer must not be null");
+        if (off < 0 || len < 0 || off + len > b.length) {
+            throw new IndexOutOfBoundsException("Invalid off/len for buffer of length " + b.length);
+        }
+        if (len == 0) {
+            return;
+        }
+
+        operationLock.lock();
+        try {
+            if (closing.get()) {
+                throw new IOException("The in-process runtime connection is closed.");
+            }
+            int id = connectionId.get();
+            if (id == 0) {
+                throw new IOException("The in-process runtime connection is closed.");
+            }
+
+            byte[] payload = (off == 0 && len == b.length) ? b : Arrays.copyOfRange(b, off, off + len);
+            if (!nativeBinding.connectionWrite(id, payload, payload.length)) {
+                throw new IOException("Failed to write a frame to the in-process runtime connection.");
+            }
+        } finally {
+            operationLock.unlock();
+        }
+    }
+}