diff --git a/lib/core/extensions/rpc/json_rpc_stdio_client.dart b/lib/core/extensions/rpc/json_rpc_stdio_client.dart
index b3f04243..e277ee2d 100644
--- a/lib/core/extensions/rpc/json_rpc_stdio_client.dart
+++ b/lib/core/extensions/rpc/json_rpc_stdio_client.dart
@@ -67,9 +67,15 @@ class JsonRpcStdioClient {
     _closed = true;
     await _subscription?.cancel();
     _subscription = null;
+    failAll(StateError('JsonRpcStdioClient closed'));
+  }
+
+  /// Completes all in-flight requests with [error] (e.g. plugin crash).
+  void failAll(Object error, [StackTrace? stackTrace]) {
+    _fatalError = error;
     for (final pending in _pending.values) {
       if (!pending.isCompleted) {
-        pending.completeError(StateError('JsonRpcStdioClient closed'));
+        pending.completeError(error, stackTrace);
       }
     }
     _pending.clear();
diff --git a/lib/core/extensions/rpc/plugin_rpc_bridge.dart b/lib/core/extensions/rpc/plugin_rpc_bridge.dart
new file mode 100644
index 00000000..89153785
--- /dev/null
+++ b/lib/core/extensions/rpc/plugin_rpc_bridge.dart
@@ -0,0 +1,242 @@
+import 'dart:async';
+
+import 'package:flutter/foundation.dart';
+import 'package:querya_desktop/core/extensions/models/extension_manifest.dart';
+import 'package:querya_desktop/core/extensions/models/sandbox_capabilities.dart';
+import 'package:querya_desktop/core/extensions/rpc/json_rpc_stdio_client.dart';
+import 'package:querya_desktop/core/extensions/rpc/plugin_rpc_exceptions.dart';
+import 'package:querya_desktop/core/extensions/sandbox/sandbox_process_runner.dart';
+import 'package:querya_desktop/core/extensions/sandbox/sandbox_security_audit.dart';
+import 'package:querya_desktop/core/extensions/sandbox/sandbox_stderr_pipe.dart';
+import 'package:querya_desktop/core/extensions/sandbox/sandbox_watchdog.dart';
+import 'package:querya_desktop/core/extensions/sandbox/sandbox_auto_recovery.dart';
+
+/// High-level JSON-RPC bridge to a sandboxed plugin process (Block C).
+///
+/// Owns process lifetime ([SandboxProcessRunner]), Stdio JSON-RPC,
+/// optional stderr sanitization, and heartbeat watchdog.
+class PluginRpcBridge {
+  PluginRpcBridge({
+    SandboxProcessRunner? processRunner,
+    this.handshakeTimeout = const Duration(seconds: 3),
+    this.shutdownTimeout = const Duration(seconds: 3),
+    this.requestTimeout = const Duration(seconds: 30),
+    this.enableWatchdog = true,
+    this.enableStderrPipe = true,
+    SandboxSecurityAudit? audit,
+    SandboxAutoRecovery? recovery,
+  })  : _runner = processRunner ?? SandboxProcessRunner(),
+        _audit = audit,
+        _recovery = recovery ?? SandboxAutoRecovery();
+
+  final SandboxProcessRunner _runner;
+  final Duration handshakeTimeout;
+  final Duration shutdownTimeout;
+  final Duration requestTimeout;
+  final bool enableWatchdog;
+  final bool enableStderrPipe;
+  final SandboxSecurityAudit? _audit;
+  final SandboxAutoRecovery _recovery;
+
+  SandboxProcessHandle? _handle;
+  JsonRpcStdioClient? _client;
+  SandboxStderrPipe? _stderrPipe;
+  SandboxWatchdog? _watchdog;
+  StreamSubscription<int>? _exitSub;
+  var _started = false;
+  var _shuttingDown = false;
+
+  bool get isStarted => _started && _handle != null && !(_handle!.isDisposed);
+
+  String? get pluginId => _handle?.pluginId;
+
+  SandboxProcessHandle? get handle => _handle;
+
+  SandboxAutoRecovery get recovery => _recovery;
+
+  /// Spawns the plugin, attaches RPC + optional pipes, and runs handshake.
+  Future<Object?> start({
+    required ExtensionManifest manifest,
+    required String pluginExecutable,
+    List<String> pluginArguments = const [],
+    String? extensionRoot,
+    Map<String, String>? environment,
+    Map<String, Object?>? handshakeParams,
+  }) async {
+    if (_started) {
+      throw StateError('PluginRpcBridge already started');
+    }
+
+    final capabilities = manifest.sandbox ??
+        const SandboxCapabilities(engine: SandboxEngine.process);
+
+    final handle = await _runner.start(
+      pluginId: manifest.id,
+      pluginExecutable: pluginExecutable,
+      pluginArguments: pluginArguments,
+      extensionRoot: extensionRoot ?? manifest.installPath,
+      capabilities: capabilities,
+      environment: environment,
+    );
+
+    _handle = handle;
+    _started = true;
+
+    final client = JsonRpcStdioClient(
+      stdout: handle.process.stdout,
+      stdin: handle.process.stdin,
+      requestTimeout: requestTimeout,
+    );
+    _client = client;
+
+    _exitSub = handle.process.exitCode.asStream().listen((code) {
+      if (!_started || _shuttingDown) return;
+      _onProcessExited(code);
+    }, onError: (Object e, StackTrace st) {
+      if (!_started || _shuttingDown) return;
+      debugPrint('PluginRpcBridge exit watch error: $e\n$st');
+      _failPending(PluginCrashedException(
+        pluginId: handle.pluginId,
+        message: '$e',
+      ));
+    });
+
+    if (enableStderrPipe) {
+      try {
+        _stderrPipe = await SandboxStderrPipe.attach(
+          handle,
+          audit: _audit,
+        );
+      } catch (e) {
+        debugPrint('PluginRpcBridge: stderr pipe attach failed: $e');
+      }
+    }
+
+    if (enableWatchdog) {
+      _watchdog = SandboxWatchdog(
+        recovery: _recovery,
+        onStopped: (reason) {
+          if (reason == SandboxWatchdogStopReason.deadlock) {
+            unawaited(_audit?.record(
+              type: SandboxSecurityEventType.deadlock,
+              pluginId: handle.pluginId,
+              detail: 'watchdog deadlock',
+            ));
+          }
+        },
+      );
+      _watchdog!.start(handle, client: client);
+    }
+
+    try {
+      final result = await client
+          .sendRequest('system.handshake', handshakeParams ?? const {})
+          .timeout(handshakeTimeout);
+      _recovery.recordSuccess();
+      return result;
+    } on TimeoutException {
+      await _forceKill();
+      throw PluginProtocolTimeoutException(
+        'system.handshake timed out after $handshakeTimeout',
+      );
+    }
+  }
+
+  /// Sends a JSON-RPC request to the plugin.
+  Future<Object?> sendRequest(String method, [Object? params]) {
+    final client = _client;
+    if (client == null || !isStarted) {
+      throw StateError('PluginRpcBridge is not started');
+    }
+    return client.sendRequest(method, params);
+  }
+
+  Future<Object?> ping() => sendRequest('system.ping');
+
+  Future<Object?> injectCredentials(Map<String, Object?> params) =>
+      sendRequest('system.injectCredentials', params);
+
+  Future<Object?> connect(Map<String, Object?> params) =>
+      sendRequest('db.connect', params);
+
+  /// Asks the plugin to shut down, then disposes the process and scratch dir.
+  Future<void> shutdown() async {
+    if (!_started && _handle == null) return;
+    _shuttingDown = true;
+    final client = _client;
+    final handle = _handle;
+
+    _watchdog?.stop();
+    _watchdog = null;
+
+    if (client != null && handle != null && !handle.isDisposed) {
+      try {
+        await client
+            .sendRequest('system.shutdown')
+            .timeout(shutdownTimeout);
+      } catch (e) {
+        debugPrint('PluginRpcBridge.shutdown RPC: $e');
+      }
+
+      try {
+        await handle.process.exitCode.timeout(shutdownTimeout);
+      } on TimeoutException {
+        await handle.kill();
+      } catch (_) {
+        await handle.kill();
+      }
+    }
+
+    await _disposeLocal();
+    _shuttingDown = false;
+  }
+
+  Future<void> _forceKill() async {
+    final handle = _handle;
+    if (handle != null && !handle.isDisposed) {
+      await handle.kill();
+    }
+    await _disposeLocal();
+  }
+
+  void _onProcessExited(int code) {
+    if (!_started) return;
+    debugPrint('PluginRpcBridge: process exited with $code');
+    if (code != 0) {
+      _recovery.recordFailure();
+    }
+    _failPending(PluginCrashedException(
+      pluginId: _handle?.pluginId ?? 'unknown',
+      exitCode: code,
+    ));
+    unawaited(_disposeLocal(keepRecovery: true));
+  }
+
+  void _failPending(Object error) {
+    _client?.failAll(error);
+    unawaited(_client?.close());
+    if (error is PluginCrashedException) {
+      debugPrint('$error');
+    }
+  }
+
+  Future<void> _disposeLocal({bool keepRecovery = false}) async {
+    _started = false;
+    await _exitSub?.cancel();
+    _exitSub = null;
+    _watchdog?.stop();
+    _watchdog = null;
+    await _stderrPipe?.close();
+    _stderrPipe = null;
+    await _client?.close();
+    _client = null;
+    final handle = _handle;
+    _handle = null;
+    if (handle != null && !handle.isDisposed) {
+      await handle.dispose();
+    }
+    if (!keepRecovery) {
+      // leave recovery state as-is for auto-restart decisions
+    }
+  }
+}
diff --git a/lib/core/extensions/rpc/plugin_rpc_exceptions.dart b/lib/core/extensions/rpc/plugin_rpc_exceptions.dart
new file mode 100644
index 00000000..257d9b4c
--- /dev/null
+++ b/lib/core/extensions/rpc/plugin_rpc_exceptions.dart
@@ -0,0 +1,28 @@
+/// Thrown when a plugin child process exits unexpectedly (Block C).
+class PluginCrashedException implements Exception {
+  PluginCrashedException({
+    required this.pluginId,
+    this.exitCode,
+    this.message,
+  });
+
+  final String pluginId;
+  final int? exitCode;
+  final String? message;
+
+  @override
+  String toString() {
+    final code = exitCode == null ? '' : ' (exitCode=$exitCode)';
+    final detail = message == null ? '' : ': $message';
+    return 'PluginCrashedException($pluginId)$code$detail';
+  }
+}
+
+/// Thrown when handshake / shutdown protocol times out.
+class PluginProtocolTimeoutException implements Exception {
+  PluginProtocolTimeoutException(this.message);
+  final String message;
+
+  @override
+  String toString() => 'PluginProtocolTimeoutException: $message';
+}
diff --git a/test/core/extensions/rpc/plugin_rpc_bridge_test.dart b/test/core/extensions/rpc/plugin_rpc_bridge_test.dart
new file mode 100644
index 00000000..aba482e2
--- /dev/null
+++ b/test/core/extensions/rpc/plugin_rpc_bridge_test.dart
@@ -0,0 +1,248 @@
+import 'dart:async';
+import 'dart:convert';
+import 'dart:io';
+
+import 'package:flutter_test/flutter_test.dart';
+import 'package:querya_desktop/core/extensions/models/extension_manifest.dart';
+import 'package:querya_desktop/core/extensions/models/extension_type.dart';
+import 'package:querya_desktop/core/extensions/models/sandbox_capabilities.dart';
+import 'package:querya_desktop/core/extensions/rpc/plugin_rpc_bridge.dart';
+import 'package:querya_desktop/core/extensions/rpc/plugin_rpc_exceptions.dart';
+import 'package:querya_desktop/core/extensions/sandbox/sandbox_process_runner.dart';
+
+class _FakeProcess implements Process {
+  _FakeProcess()
+      : _stdoutController = StreamController<List<int>>.broadcast(),
+        _stderrController = StreamController<List<int>>.broadcast(),
+        _stdinController = StreamController<List<int>>() {
+    stdin = IOSink(_stdinController.sink);
+    stdinLines = utf8.decoder
+        .bind(_stdinController.stream)
+        .transform(const LineSplitter())
+        .asBroadcastStream();
+  }
+
+  final StreamController<List<int>> _stdoutController;
+  final StreamController<List<int>> _stderrController;
+  final StreamController<List<int>> _stdinController;
+  final _exit = Completer<int>();
+
+  late final Stream<String> stdinLines;
+
+  @override
+  int get pid => 99;
+
+  @override
+  late final IOSink stdin;
+
+  @override
+  Stream<List<int>> get stdout => _stdoutController.stream;
+
+  @override
+  Stream<List<int>> get stderr => _stderrController.stream;
+
+  @override
+  Future<int> get exitCode => _exit.future;
+
+  @override
+  bool kill([ProcessSignal signal = ProcessSignal.sigterm]) {
+    if (!_exit.isCompleted) {
+      _exit.complete(signal == ProcessSignal.sigkill ? -9 : 0);
+    }
+    return true;
+  }
+
+  void reply(Map<String, Object?> message) {
+    _stdoutController.add(utf8.encode('${jsonEncode(message)}\n'));
+  }
+
+  void completeExit([int code = 0]) {
+    if (!_exit.isCompleted) _exit.complete(code);
+  }
+}
+
+void main() {
+  late Directory tempBase;
+
+  setUp(() async {
+    tempBase = await Directory.systemTemp.createTemp('querya_rpc_bridge_');
+  });
+
+  tearDown(() async {
+    try {
+      if (await tempBase.exists()) {
+        await tempBase.delete(recursive: true);
+      }
+    } on PathNotFoundException {
+      // Already cleaned by process dispose races.
+    } on FileSystemException {
+      // Best-effort cleanup.
+    }
+  });
+
+  const testManifest = ExtensionManifest(
+    id: 'test.rpc-driver',
+    name: 'RPC Driver',
+    version: '1.0.0',
+    publisher: 'Test',
+    type: ExtensionType.databaseDriver,
+    engines: {'querya_desktop': '*'},
+    main: 'bin/driver',
+    sandbox: SandboxCapabilities(
+      engine: SandboxEngine.process,
+      network: NetworkPermission(
+        mode: NetworkPermissionMode.connectionHostOnly,
+      ),
+    ),
+  );
+
+  test('start performs handshake and sendRequest works', () async {
+    final process = _FakeProcess();
+    final sub = process.stdinLines.listen((line) {
+      final req = jsonDecode(line) as Map<String, dynamic>;
+      final method = req['method'];
+      if (method == 'system.handshake') {
+        process.reply({
+          'jsonrpc': '2.0',
+          'id': req['id'] as int,
+          'result': {
+            'protocolVersion': '1.0',
+            'capabilities': ['db.connect'],
+          },
+        });
+      } else if (method == 'db.connect') {
+        process.reply({
+          'jsonrpc': '2.0',
+          'id': req['id'] as int,
+          'result': {'ok': true},
+        });
+      } else if (method == 'system.shutdown') {
+        process.reply({
+          'jsonrpc': '2.0',
+          'id': req['id'] as int,
+          'result': null,
+        });
+        process.completeExit(0);
+      }
+    });
+
+    final runner = SandboxProcessRunner(
+      platformOverride: 'windows',
+      scratchBaseDirectory: tempBase,
+      processStarter: (
+        exe,
+        args, {
+        String? workingDirectory,
+        Map<String, String>? environment,
+        bool includeParentEnvironment = true,
+        bool runInShell = false,
+        ProcessStartMode mode = ProcessStartMode.normal,
+      }) async =>
+          process,
+    );
+
+    final bridge = PluginRpcBridge(
+      processRunner: runner,
+      enableWatchdog: false,
+      enableStderrPipe: false,
+    );
+
+    final handshake = await bridge.start(
+      manifest: testManifest,
+      pluginExecutable: '/opt/driver',
+    );
+    expect(handshake, isA<Map>());
+    expect((handshake as Map)['protocolVersion'], '1.0');
+
+    final connected = await bridge.connect({'host': 'localhost', 'port': 5432});
+    expect(connected, {'ok': true});
+
+    await bridge.shutdown();
+    expect(bridge.isStarted, isFalse);
+    await sub.cancel();
+  });
+
+  test('handshake timeout kills process', () async {
+    final process = _FakeProcess();
+    // Never reply to handshake.
+    final sub = process.stdinLines.listen((_) {});
+
+    final runner = SandboxProcessRunner(
+      platformOverride: 'windows',
+      scratchBaseDirectory: tempBase,
+      processStarter: (
+        exe,
+        args, {
+        String? workingDirectory,
+        Map<String, String>? environment,
+        bool includeParentEnvironment = true,
+        bool runInShell = false,
+        ProcessStartMode mode = ProcessStartMode.normal,
+      }) async =>
+          process,
+    );
+
+    final bridge = PluginRpcBridge(
+      processRunner: runner,
+      handshakeTimeout: const Duration(milliseconds: 40),
+      enableWatchdog: false,
+      enableStderrPipe: false,
+    );
+
+    await expectLater(
+      bridge.start(manifest: testManifest, pluginExecutable: '/opt/driver'),
+      throwsA(isA<PluginProtocolTimeoutException>()),
+    );
+    expect(bridge.isStarted, isFalse);
+    await sub.cancel();
+  });
+
+  test('unexpected exit fails in-flight requests with PluginCrashedException',
+      () async {
+    final process = _FakeProcess();
+    final sub = process.stdinLines.listen((line) {
+      final req = jsonDecode(line) as Map<String, dynamic>;
+      if (req['method'] == 'system.handshake') {
+        process.reply({
+          'jsonrpc': '2.0',
+          'id': req['id'] as int,
+          'result': {'ok': true},
+        });
+      }
+      // Leave db.connect hanging until crash.
+    });
+
+    final runner = SandboxProcessRunner(
+      platformOverride: 'windows',
+      scratchBaseDirectory: tempBase,
+      processStarter: (
+        exe,
+        args, {
+        String? workingDirectory,
+        Map<String, String>? environment,
+        bool includeParentEnvironment = true,
+        bool runInShell = false,
+        ProcessStartMode mode = ProcessStartMode.normal,
+      }) async =>
+          process,
+    );
+
+    final bridge = PluginRpcBridge(
+      processRunner: runner,
+      enableWatchdog: false,
+      enableStderrPipe: false,
+      requestTimeout: const Duration(seconds: 5),
+    );
+
+    await bridge.start(manifest: testManifest, pluginExecutable: '/opt/driver');
+    final pending = bridge.connect({'host': 'x'});
+    await Future<void>.delayed(const Duration(milliseconds: 20));
+    process.completeExit(1);
+
+    await expectLater(
+      pending,
+      throwsA(isA<PluginCrashedException>().having((e) => e.exitCode, 'code', 1)),
+    );
+    await sub.cancel();
+  });
+}