diff --git a/architecture/notes/nebula_app/ssh_session/2026-09-26-pane-local-forwarding.md b/architecture/notes/nebula_app/ssh_session/2026-09-26-pane-local-forwarding.md
new file mode 100644
index 00000000..dc82e3f3
--- /dev/null
+++ b/architecture/notes/nebula_app/ssh_session/2026-09-26-pane-local-forwarding.md
@@ -0,0 +1,78 @@
+# Pane-owned local SSH forwarding
+
+## Status
+
+Reviewed for integration on 2026-09-26.
+
+## Context
+
+An authenticated native SSH pane needs a small local forwarding control. Opening
+another SSH process would duplicate authentication and connection ownership.
+A listener must not survive its pane, including normal shell exit, and a pending
+creation must not finish after the owner has closed.
+
+## Evidence
+
+- `ssh_session/forward.rs` reuses authenticated pooled sessions and opens
+  direct-tcpip channels through the existing SSH runtime.
+- The original change cleared established forwards on failure but not normal
+  exit; its detached creation task outlived the pane.
+- An unrestricted accept loop could create arbitrary numbers of channel tasks
+  and bidirectional copy buffers from local clients.
+- The legacy dialog child scroller clipped the port controls: the real-click
+  regression left the port value unchanged after typing. The form now uses
+  `DialogContent`, with explicit input geometry and read-back assertions.
+- The original screenshot-only readiness override read the process environment
+  from a render-time query and bypassed real SSH readiness in test builds.
+
+## Decision
+
+Keep listeners in the terminal view and the pending GPUI operation in an owned
+Task. Exit, connection failure and view destruction release both. The existing
+network-to-GPUI bridge cancels its future when the result receiver is dropped;
+a stale completion cannot install a forward or show a toast for a dead pane.
+
+Each listener accepts at most 64 concurrent channel tasks. Excess clients wait
+in the OS TCP backlog until a slot is available. This bounds task/copy-buffer
+cost per listener; it is an engineering limit, not a throughput guarantee.
+Completed channels are reaped and failures are logged without killing the SSH
+transport shared by the terminal and other forwards.
+
+The endpoints remain loopback-only on both sides. Creating and stopping a
+forward requires explicit user action. New labels use typed catalog messages.
+Readiness uses actual session state; screenshots cannot enable SSH on local panes.
+
+## Rejected alternatives
+
+- A second SSH process or global forwarding service would duplicate lifetime
+  and authentication policy.
+- Detached creation cannot guarantee cancellation when the pane disappears.
+- Unbounded per-client tasks allow local load to grow memory without a limit.
+- Persisted rules, automatic discovery and remote/dynamic forwarding are outside
+  this local forwarding capability.
+
+## Consequences
+
+The new behavior adds no persistent thread or timer. UI rendering does not read
+files, resolve destinations or inspect environment variables. File/profile
+resolution and network I/O remain in the existing background runtime. A forward
+owns its channels but never disconnects the shared authenticated transport.
+
+## Validation
+
+Regression coverage exercises real SSH channel exchange, half-close, occupied
+ports, channel rejection, listener/connection disposal, and bounded concurrency.
+UI regressions click the actual port fields and confirmation/cancel controls,
+check invalid input, and reject submission after the pane stops being ready.
+Terminal and bridge regressions cover normal exit, failure and pending-work
+cancellation. These tests do not claim arbitrary-host throughput or physical
+platform visual acceptance.
+
+## Supersedes
+
+None.
+
+## Revisit when
+
+Revisit the limit and UI state model if measured workloads need more concurrent
+channels, or when persistent/reconnecting forwards are explicitly introduced.
diff --git a/nebula_app/i18n/en-US.json b/nebula_app/i18n/en-US.json
index 4b9a99d3..b47d9453 100644
--- a/nebula_app/i18n/en-US.json
+++ b/nebula_app/i18n/en-US.json
@@ -1000,5 +1000,16 @@
     "empty": "This directory is empty.",
     "error_retry": "{error} (click Reload in the top-right to retry)",
     "busy_other": "Another host is transferring: {destination}"
-  }
+  },
+  "ssh.ports.button": "Ports",
+  "ssh.ports.title": "SSH port forwarding",
+  "ssh.ports.empty": "No forwarded ports",
+  "ssh.ports.stop": "Stop forwarding",
+  "ssh.ports.remote": "Remote port",
+  "ssh.ports.local": "Local port",
+  "ssh.ports.forward": "Forward",
+  "ssh.ports.invalid": "Enter ports from 1–65535",
+  "ssh.ports.pending": "Starting port forwarding…",
+  "ssh.ports.unavailable": "SSH network runtime unavailable",
+  "ssh.ports.started": "Forwarding 127.0.0.1:{local} to {host} port {remote}"
 }
diff --git a/nebula_app/i18n/zh-CN.json b/nebula_app/i18n/zh-CN.json
index a78fd32e..4a112b61 100644
--- a/nebula_app/i18n/zh-CN.json
+++ b/nebula_app/i18n/zh-CN.json
@@ -1000,5 +1000,16 @@
     "empty": "此目录为空。",
     "error_retry": "{error}（点右上角重新读取）",
     "busy_other": "另一主机正在传输：{destination}"
-  }
+  },
+  "ssh.ports.button": "端口",
+  "ssh.ports.title": "SSH 端口转发",
+  "ssh.ports.empty": "暂无端口转发",
+  "ssh.ports.stop": "停止转发",
+  "ssh.ports.remote": "服务器端口",
+  "ssh.ports.local": "本地端口",
+  "ssh.ports.forward": "转发",
+  "ssh.ports.invalid": "请输入 1–65535 的端口",
+  "ssh.ports.pending": "正在启动端口转发…",
+  "ssh.ports.unavailable": "SSH 网络运行时不可用",
+  "ssh.ports.started": "正在将 127.0.0.1:{local} 转发至 {host} 的 {remote} 端口"
 }
diff --git a/nebula_app/src/gpui_shell/terminal/view.rs b/nebula_app/src/gpui_shell/terminal/view.rs
index eddb5c6f..24ff8df0 100644
--- a/nebula_app/src/gpui_shell/terminal/view.rs
+++ b/nebula_app/src/gpui_shell/terminal/view.rs
@@ -336,6 +336,9 @@ pub struct TerminalView {
     path_drop: path_drop::PathDropState,
     /// SSH 直连目的地（`user@host[:port]`）；本地会话为 None。
     pub ssh_destination: Option<String>,
+    /// 本 pane 拥有的本地端口转发；pane 销毁即停止监听。
+    pub(crate) port_forwards: Vec<crate::ssh_session::LocalForward>,
+    pub(crate) port_forward_task: Option<gpui::Task<()>>,
     ssh_label: Option<String>,
     /// 创建本地 PTY 时冻结的受控环境，供独立 `pane.exec` child 复用。
     pub(crate) exec_context: Option<crate::runtime_exec::PaneExecContext>,
@@ -706,6 +709,8 @@ impl TerminalView {
 
     /// `Exited` 只对宿主发一次；重复的退出信号（ChildExit 之后必然跟 Exit）只更新文案。
     fn mark_exited(&mut self, message: String, cx: &mut Context<Self>) {
+        self.port_forward_task = None;
+        self.port_forwards.clear();
         self.confirmation.invalidate();
         self.pending_runtime_submit = None;
         self.pending_shell_command = None;
@@ -971,6 +976,9 @@ impl TerminalView {
     /// 通道，撞上一个还没建立的传输——用户看到的是文件面板先报一个错，然后
     /// 终端才连上。
     pub fn ready_ssh_destination(&self) -> Option<&str> {
+        if self.exited.is_some() {
+            return None;
+        }
         let destination = self.ssh_destination.as_deref()?;
         matches!(self.ssh_stage, Some(crate::ssh_session::SshStage::Ready)).then_some(destination)
     }
diff --git a/nebula_app/src/gpui_shell/terminal/view/output_tests.rs b/nebula_app/src/gpui_shell/terminal/view/output_tests.rs
index 595bf150..c6ba8cc2 100644
--- a/nebula_app/src/gpui_shell/terminal/view/output_tests.rs
+++ b/nebula_app/src/gpui_shell/terminal/view/output_tests.rs
@@ -177,3 +177,28 @@ fn hidden_wakeup_still_flushes_shell_commands_and_pending_enter(cx: &mut TestApp
         assert!(view.pending_runtime_submit.is_none());
     });
 }
+
+#[gpui::test]
+fn ssh_exit_and_failure_cancel_pending_forward_and_hide_remote_actions(cx: &mut TestAppContext) {
+    let (view, window, _) = open(cx);
+    for failed in [false, true] {
+        view.update(window, |view, cx| {
+            view.exited = None;
+            view.ssh_destination = Some("fixture@localhost".into());
+            view.apply_ssh_stage(crate::ssh_session::SshStage::Ready, cx);
+            assert_eq!(view.ready_ssh_destination(), Some("fixture@localhost"));
+            view.port_forward_task =
+                Some(cx.spawn(async |_, _| std::future::pending::<()>().await));
+            if failed {
+                view.apply_ssh_stage(
+                    crate::ssh_session::SshStage::Failed("disconnected".into()),
+                    cx,
+                );
+            } else {
+                view.process_event(TermEvent::Exit, cx);
+            }
+            assert!(view.port_forward_task.is_none());
+            assert!(view.ready_ssh_destination().is_none());
+        });
+    }
+}
diff --git a/nebula_app/src/gpui_shell/terminal/view/runtime.rs b/nebula_app/src/gpui_shell/terminal/view/runtime.rs
index 19402253..ca32c59b 100644
--- a/nebula_app/src/gpui_shell/terminal/view/runtime.rs
+++ b/nebula_app/src/gpui_shell/terminal/view/runtime.rs
@@ -111,6 +111,8 @@ impl TerminalView {
         );
         self.ssh_connect_last_step = std::time::Instant::now();
         if matches!(stage, crate::ssh_session::SshStage::Failed(_)) {
+            self.port_forward_task = None;
+            self.port_forwards.clear();
             self.pending_runtime_submit = None;
             self.pending_shell_command = None;
             self.command_running = false;
diff --git a/nebula_app/src/gpui_shell/terminal/view/startup.rs b/nebula_app/src/gpui_shell/terminal/view/startup.rs
index 58167db2..f21b497a 100644
--- a/nebula_app/src/gpui_shell/terminal/view/startup.rs
+++ b/nebula_app/src/gpui_shell/terminal/view/startup.rs
@@ -284,6 +284,8 @@ impl TerminalView {
             image_paste: image_paste::ImagePasteState::default(),
             path_drop: path_drop::PathDropState::default(),
             ssh_destination,
+            port_forwards: Vec::new(),
+            port_forward_task: None,
             ssh_label: None,
             exec_context,
             ssh_stage: None,
diff --git a/nebula_app/src/gpui_shell/workspace/remote_files.rs b/nebula_app/src/gpui_shell/workspace/remote_files.rs
index 0b412b15..1fca94ec 100644
--- a/nebula_app/src/gpui_shell/workspace/remote_files.rs
+++ b/nebula_app/src/gpui_shell/workspace/remote_files.rs
@@ -920,16 +920,20 @@ fn format_transfer_bytes(bytes: u64) -> String {
 ///
 /// 返回 `None` 表示网络 runtime 起不来或任务被丢弃——调用方据此报"连接不
 /// 可用"，而不是把它和"远端答了个错误"混为一谈。
-async fn remote_call<T, F, Fut>(work: F) -> Option<T>
+pub(super) async fn remote_call<T, F, Fut>(work: F) -> Option<T>
 where
     T: Send + 'static,
     F: FnOnce() -> Fut + Send + 'static,
     Fut: std::future::Future<Output = T> + Send,
 {
     let runtime = crate::ssh_session::runtime().ok()?;
-    let (tx, rx) = tokio::sync::oneshot::channel();
+    let (mut tx, rx) = tokio::sync::oneshot::channel();
     runtime.spawn(async move {
-        let _ = tx.send(work().await);
+        tokio::select! {
+            biased;
+            _ = tx.closed() => {},
+            result = work() => { let _ = tx.send(result); },
+        }
     });
     rx.await.ok()
 }
@@ -950,6 +954,37 @@ mod tests {
         }
     }
 
+    #[test]
+    fn dropped_remote_call_cancels_its_network_work() {
+        struct SignalDrop(Option<tokio::sync::oneshot::Sender<()>>);
+        impl Drop for SignalDrop {
+            fn drop(&mut self) {
+                if let Some(sender) = self.0.take() {
+                    let _ = sender.send(());
+                }
+            }
+        }
+        crate::ssh_session::runtime().unwrap().block_on(async {
+            let (started_tx, started_rx) = tokio::sync::oneshot::channel();
+            let (dropped_tx, dropped_rx) = tokio::sync::oneshot::channel();
+            let caller = tokio::spawn(super::remote_call(move || async move {
+                let _signal = SignalDrop(Some(dropped_tx));
+                let _ = started_tx.send(());
+                std::future::pending::<()>().await;
+            }));
+            tokio::time::timeout(std::time::Duration::from_secs(5), started_rx)
+                .await
+                .unwrap()
+                .unwrap();
+            caller.abort();
+            let _ = caller.await;
+            tokio::time::timeout(std::time::Duration::from_secs(5), dropped_rx)
+                .await
+                .unwrap()
+                .unwrap();
+        });
+    }
+
     #[test]
     fn switching_remote_panes_restores_listing_selection_and_scroll_owner_without_reload() {
         let mut browser = RemoteBrowser::default();
diff --git a/nebula_app/src/gpui_shell/workspace/sidebar.rs b/nebula_app/src/gpui_shell/workspace/sidebar.rs
index 8d6074a4..b9d98576 100644
--- a/nebula_app/src/gpui_shell/workspace/sidebar.rs
+++ b/nebula_app/src/gpui_shell/workspace/sidebar.rs
@@ -868,6 +868,7 @@ impl NebulaWorkspace {
         let settings_active_fg = cx.theme().sidebar_accent_foreground;
         let sidebar_visible = !self.sidebar_collapsed && !self.reader_focus_active(cx);
         let language = crate::gpui_shell::config::ui_language(cx);
+        let port_forward_button = self.render_port_forward_button(cx);
         h_flex()
             .size_full()
             .items_center()
@@ -932,6 +933,7 @@ impl NebulaWorkspace {
                                 this.toggle_command_manager(window, cx);
                             })),
                     )
+                    .when_some(port_forward_button, |controls, button| controls.child(button))
                     .child(self.render_right_sidebar_button(settings_active, cx)),
             )
             .into_any_element()
diff --git a/nebula_app/src/gpui_shell/workspace/ssh_dialog.rs b/nebula_app/src/gpui_shell/workspace/ssh_dialog.rs
index fa89fd15..163c6c1d 100644
--- a/nebula_app/src/gpui_shell/workspace/ssh_dialog.rs
+++ b/nebula_app/src/gpui_shell/workspace/ssh_dialog.rs
@@ -2,11 +2,18 @@ use std::cell::Cell;
 use std::rc::Rc;
 use std::sync::Arc;
 
-use gpui::{App, AppContext as _, ParentElement as _, Styled as _, Window, div};
+use gpui::{
+    App, AppContext as _, Context, Entity, InteractiveElement as _, ParentElement as _,
+    Styled as _, Window, div,
+};
 
 use crate::gpui_shell::prelude::*;
+use crate::gpui_shell::terminal::view::TerminalView;
+use crate::i18n::Message;
 use crate::ssh_prompt::{Prompt, PromptKind, PromptResponse};
 
+use super::{NebulaWorkspace, workspace_ui_language};
+
 pub(super) fn show(request: Arc<Prompt>, window: &mut Window, cx: &mut App) {
     if !request.is_pending() {
         return;
@@ -176,4 +183,359 @@ mod tests {
             _ => panic!("confirm must deliver the entered password"),
         }
     }
+    #[gpui::test]
+    fn ports_dialog_rejects_invalid_input_and_cancel_releases_focus(cx: &mut TestAppContext) {
+        cx.update(|cx| {
+            gpui_component::init(cx);
+            cx.set_reduce_motion(true);
+        });
+        let (_, cx) =
+            cx.add_window_view(|window, cx| Root::new(cx.new(|_| DialogProbe), window, cx));
+        let owner = cx.update(|window, cx| {
+            cx.new(|cx| {
+                TerminalView::new(
+                    42,
+                    (80, 24),
+                    crate::gpui_shell::terminal::view::TerminalLaunch::Local {
+                        cwd: None,
+                        shell: Some(nebula_terminal::tty::Shell::new(
+                            "pebrel-test-missing-shell-executable".into(),
+                            vec![],
+                        )),
+                        shell_name: None,
+                    },
+                    window,
+                    cx,
+                )
+            })
+        });
+        let inputs = cx.update(|window, cx| {
+            show_port_forward_dialog(owner.clone(), "fixture@localhost".into(), window, cx)
+        });
+        cx.update(|window, cx| {
+            window.refresh();
+            window.draw(cx).clear(cx);
+        });
+        cx.simulate_input("65536");
+        assert_eq!(inputs[0].read_with(cx, |input, _| input.value()).as_ref(), "65536");
+        click(cx, "confirm-dialog-ok");
+        cx.update(|window, cx| {
+            window.refresh();
+            window.draw(cx).clear(cx);
+        });
+        assert!(
+            cx.debug_bounds("confirm-dialog-ok").is_some(),
+            "invalid input keeps the dialog open"
+        );
+        owner.read_with(cx, |view, _| {
+            assert!(view.port_forward_task.is_none());
+            assert!(view.port_forwards.is_empty());
+        });
+        let select_all = match crate::platform::Platform::current() {
+            crate::platform::Platform::MacOS => "cmd-a",
+            _ => "ctrl-a",
+        };
+        for (index, selector) in ["ssh-forward-remote", "ssh-forward-local"].into_iter().enumerate()
+        {
+            let bounds = cx.debug_bounds(selector).unwrap();
+            assert!(f32::from(bounds.size.height) >= 28.0);
+            assert!(f32::from(bounds.size.width) >= 100.0);
+            click(cx, selector);
+            cx.simulate_keystrokes(select_all);
+            cx.simulate_input("3000");
+            assert_eq!(
+                inputs[index].read_with(cx, |input, _| input.value()).as_ref(),
+                "3000",
+                "clicking {selector} must focus and edit that field"
+            );
+        }
+        click(cx, "confirm-dialog-ok");
+        cx.update(|window, cx| {
+            window.refresh();
+            window.draw(cx).clear(cx);
+        });
+        assert!(
+            cx.debug_bounds("confirm-dialog-ok").is_none(),
+            "valid ports close the dialog when its pane is no longer ready"
+        );
+        owner.read_with(cx, |view, _| assert!(view.port_forward_task.is_none()));
+        cx.update(|window, cx| {
+            show_port_forward_dialog(owner.clone(), "fixture@localhost".into(), window, cx)
+        });
+        cx.update(|window, cx| {
+            window.refresh();
+            window.draw(cx).clear(cx);
+        });
+        click(cx, "confirm-dialog-cancel");
+        cx.update(|window, cx| {
+            window.refresh();
+            window.draw(cx).clear(cx);
+        });
+        assert!(cx.debug_bounds("confirm-dialog-ok").is_none());
+    }
+}
+
+impl NebulaWorkspace {
+    fn active_ready_ssh_view(&self, cx: &App) -> Option<Entity<TerminalView>> {
+        if self.settings_open {
+            return None;
+        }
+        let view = self.tabs.get(self.active)?.focused_view()?.clone();
+        view.read(cx).ready_ssh_destination()?;
+        Some(view)
+    }
+
+    pub(super) fn render_port_forward_button(&self, cx: &mut Context<Self>) -> Option<Button> {
+        let view = self.active_ready_ssh_view(cx)?;
+        let pending = view.read(cx).port_forward_task.is_some();
+        let language = workspace_ui_language();
+        Some(
+            Button::new("ssh-ports")
+                .icon(IconName::Network)
+                .ghost()
+                .disabled(pending)
+                .tooltip(language.text(if pending {
+                    Message::SshPortsPending
+                } else {
+                    Message::SshPortsButton
+                }))
+                .on_click(cx.listener(|this, _, window, cx| {
+                    this.open_port_forward_dialog(window, cx);
+                })),
+        )
+    }
+
+    fn open_port_forward_dialog(&mut self, window: &mut Window, cx: &mut Context<Self>) {
+        let Some(view) = self.active_ready_ssh_view(cx) else { return };
+        if view.read(cx).port_forward_task.is_some() {
+            return;
+        }
+        let destination = view.read(cx).ready_ssh_destination().unwrap().to_owned();
+        show_port_forward_dialog(view, destination, window, cx);
+    }
+}
+
+fn show_port_forward_dialog(
+    view: Entity<TerminalView>,
+    destination: String,
+    window: &mut Window,
+    cx: &mut App,
+) -> [Entity<InputState>; 2] {
+    let forwards = Rc::new(
+        view.read(cx)
+            .port_forwards
+            .iter()
+            .map(|forward| (forward.local_port(), forward.remote_port()))
+            .collect::<Vec<_>>(),
+    );
+    let remote_input = cx.new(|cx| InputState::new(window, cx).placeholder("3000"));
+    let local_input = cx.new(|cx| InputState::new(window, cx).placeholder("3000"));
+    let focus_input = remote_input.clone();
+    let inputs = [remote_input.clone(), local_input.clone()];
+    let target = view.downgrade();
+    let language = workspace_ui_language();
+
+    window.open_dialog(cx, move |dialog, window, _cx| {
+        let submit_target = target.clone();
+        let submit_remote = remote_input.clone();
+        let submit_local = local_input.clone();
+        let submit_destination = destination.clone();
+        let content_target = target.clone();
+        let content_forwards = forwards.clone();
+        let content_remote = remote_input.clone();
+        let content_local = local_input.clone();
+        let content_destination = destination.clone();
+        confirm_dialog(
+            dialog,
+            window,
+            language.text(Message::SshPortsTitle),
+            destination.clone(),
+            language.text(Message::SshPortsForward),
+            language.text(Message::CommonCancel),
+            ButtonVariant::Primary,
+        )
+        .content(move |content, _, _cx| {
+            let mut body = v_flex().w_full().gap_3();
+            if content_forwards.is_empty() {
+                body = body.child(
+                    div()
+                        .text_sm()
+                        .text_color(_cx.theme().muted_foreground)
+                        .child(language.text(Message::SshPortsEmpty)),
+                );
+            } else {
+                for (index, (local_port, remote_port)) in
+                    content_forwards.iter().copied().enumerate()
+                {
+                    let stop_target = content_target.clone();
+                    body = body.child(
+                        h_flex()
+                            .w_full()
+                            .items_center()
+                            .justify_between()
+                            .child(format!("127.0.0.1:{local_port} → 127.0.0.1:{remote_port}"))
+                            .child(
+                                Button::new(("ssh-port-stop", index))
+                                    .icon(IconName::Close)
+                                    .ghost()
+                                    .xsmall()
+                                    .tooltip(language.text(Message::SshPortsStop))
+                                    .on_click(move |_, window, cx| {
+                                        if let Some(view) = stop_target.upgrade() {
+                                            view.update(cx, |view, cx| {
+                                                if let Some(position) =
+                                                    view.port_forwards.iter().position(|forward| {
+                                                        forward.local_port() == local_port
+                                                    })
+                                                {
+                                                    view.port_forwards.remove(position);
+                                                    cx.notify();
+                                                }
+                                            });
+                                        }
+                                        window.close_dialog(cx);
+                                    }),
+                            ),
+                    );
+                }
+            }
+            body = body.child(
+                h_flex()
+                    .w_full()
+                    .gap_3()
+                    .child(
+                        v_flex()
+                            .flex_1()
+                            .gap_1()
+                            .child(language.text(Message::SshPortsRemote))
+                            .child(
+                                div()
+                                    .debug_selector(|| "ssh-forward-remote".to_owned())
+                                    .w_full()
+                                    .h_8()
+                                    .flex_shrink_0()
+                                    .child(Input::new(&content_remote).w_full()),
+                            ),
+                    )
+                    .child(
+                        v_flex()
+                            .flex_1()
+                            .gap_1()
+                            .child(language.text(Message::SshPortsLocal))
+                            .child(
+                                div()
+                                    .debug_selector(|| "ssh-forward-local".to_owned())
+                                    .w_full()
+                                    .h_8()
+                                    .flex_shrink_0()
+                                    .child(Input::new(&content_local).w_full()),
+                            ),
+                    ),
+            );
+            content
+                .child(
+                    gpui_component::dialog::DialogDescription::new()
+                        .child(content_destination.clone()),
+                )
+                .child(body)
+        })
+        .on_ok(move |_, window, cx| {
+            let remote_port =
+                submit_remote.read(cx).value().trim().parse::<u16>().ok().filter(|port| *port != 0);
+            let local_port =
+                submit_local.read(cx).value().trim().parse::<u16>().ok().filter(|port| *port != 0);
+            let (Some(remote_port), Some(local_port)) = (remote_port, local_port) else {
+                crate::gpui_shell::toast::toast(
+                    window,
+                    cx,
+                    crate::display::ToastKind::Warning,
+                    language.text(Message::SshPortsInvalid),
+                );
+                return false;
+            };
+
+            let Some(owner) = submit_target.upgrade() else {
+                return true;
+            };
+            if owner.read(cx).ready_ssh_destination() != Some(submit_destination.as_str()) {
+                return true;
+            }
+            if owner.read(cx).port_forward_task.is_some() {
+                return false;
+            }
+            let weak = submit_target.clone();
+            let expected_destination = submit_destination.clone();
+            let work_destination = submit_destination.clone();
+            let window_handle = window.window_handle();
+            let task = cx.spawn(async move |cx| {
+                let result = super::remote_files::remote_call(move || async move {
+                    crate::ssh_session::open_local_forward(
+                        &work_destination,
+                        local_port,
+                        remote_port,
+                    )
+                    .await
+                })
+                .await;
+                let _ = window_handle.update(cx, move |_, window, cx| {
+                    let Some(view) = weak.upgrade() else {
+                        return;
+                    };
+                    if view.read(cx).ready_ssh_destination() != Some(expected_destination.as_str())
+                    {
+                        return;
+                    }
+                    view.update(cx, |view, cx| {
+                        view.port_forward_task = None;
+                        cx.notify();
+                    });
+                    match result {
+                        Some(Ok(forward)) => {
+                            if let Some(view) = weak.upgrade()
+                                && view.read(cx).ready_ssh_destination()
+                                    == Some(expected_destination.as_str())
+                            {
+                                view.update(cx, |view, cx| {
+                                    view.port_forwards.push(forward);
+                                    cx.notify();
+                                });
+                                crate::gpui_shell::toast::toast(
+                                    window,
+                                    cx,
+                                    crate::display::ToastKind::Info,
+                                    language.format(
+                                        Message::SshPortsStarted,
+                                        &[
+                                            ("local", &local_port.to_string()),
+                                            ("host", &expected_destination),
+                                            ("remote", &remote_port.to_string()),
+                                        ],
+                                    ),
+                                );
+                            }
+                        },
+                        Some(Err(error)) => crate::gpui_shell::toast::toast(
+                            window,
+                            cx,
+                            crate::display::ToastKind::Warning,
+                            error.to_string(),
+                        ),
+                        None => crate::gpui_shell::toast::toast(
+                            window,
+                            cx,
+                            crate::display::ToastKind::Warning,
+                            language.text(Message::SshPortsUnavailable),
+                        ),
+                    }
+                });
+            });
+            owner.update(cx, |view, cx| {
+                view.port_forward_task = Some(task);
+                cx.notify();
+            });
+            true
+        })
+    });
+    focus_input.update(cx, |input, cx| input.focus(window, cx));
+    inputs
 }
diff --git a/nebula_app/src/gpui_shell/workspace/top_tabs.rs b/nebula_app/src/gpui_shell/workspace/top_tabs.rs
index 73453844..f0c35a3f 100644
--- a/nebula_app/src/gpui_shell/workspace/top_tabs.rs
+++ b/nebula_app/src/gpui_shell/workspace/top_tabs.rs
@@ -131,6 +131,7 @@ impl NebulaWorkspace {
         cx: &mut Context<Self>,
     ) -> gpui::AnyElement {
         let language = crate::gpui_shell::config::ui_language(cx);
+        let port_forward_button = self.render_port_forward_button(cx);
         let theme = cx.theme();
         let muted = theme.muted_foreground;
         let active_bg = theme.sidebar_accent;
@@ -145,8 +146,10 @@ impl NebulaWorkspace {
         let chrome_family = theme.mono_font_family.clone();
         let symbol_family: SharedString = crate::font_install::REQUIRED_FONT_FAMILY.into();
         let label_px = settings.map(|settings| settings.ui_font_size_px).unwrap_or(15.0);
+        let port_forward_w = if port_forward_button.is_some() { 32.0 } else { 0.0 };
         let tab_capacity_w =
-            (f32::from(window.viewport_size().width) - TOP_TAB_RESERVED_W).max(TOP_TAB_MIN_W);
+            (f32::from(window.viewport_size().width) - TOP_TAB_RESERVED_W - port_forward_w)
+                .max(TOP_TAB_MIN_W);
         let tab_w = tab_width(tab_capacity_w, self.top_tab_count());
         let strip_w = tab_strip_width(tab_w, self.top_tab_count());
         // 溢出时两端各让出一枚翻页按钮。这个反馈是单调的：`tab_w` 已被
@@ -700,6 +703,7 @@ impl NebulaWorkspace {
                                 this.toggle_command_manager(window, cx);
                             })),
                     )
+                    .when_some(port_forward_button, |controls, button| controls.child(button))
                     .child(self.render_right_sidebar_button(settings_active, cx)),
             )
             .into_any_element()
diff --git a/nebula_app/src/ssh_session.rs b/nebula_app/src/ssh_session.rs
index 4b8761b0..5a757c03 100644
--- a/nebula_app/src/ssh_session.rs
+++ b/nebula_app/src/ssh_session.rs
@@ -28,9 +28,11 @@ use crate::proxy_test::{ProxyTestFailure, ProxyTestOutcome, ProxyTestResult, Pro
 mod agent;
 mod config;
 mod exec;
+mod forward;
 mod integration;
 mod lifecycle;
 mod route;
+pub(crate) use forward::{LocalForward, open_local_forward};
 pub(crate) use integration::setup_cli as setup_ai_cli;
 use route::{ResolvedRoute, RouteTransport};
 
diff --git a/nebula_app/src/ssh_session/forward.rs b/nebula_app/src/ssh_session/forward.rs
new file mode 100644
index 00000000..1db7cd05
--- /dev/null
+++ b/nebula_app/src/ssh_session/forward.rs
@@ -0,0 +1,97 @@
+use std::net::Ipv4Addr;
+
+use tokio::net::TcpListener;
+use tokio::task::{JoinHandle, JoinSet};
+
+use super::{NoopSshEventHost, SessionError, SharedSession, SshDestination, authenticated_session};
+
+// Bound per-listener channel tasks and copy buffers; excess clients wait in the TCP backlog.
+const MAX_CONNECTIONS: usize = 64;
+
+pub(crate) struct LocalForward {
+    local_port: u16,
+    remote_port: u16,
+    task: JoinHandle<()>,
+}
+
+impl LocalForward {
+    pub(crate) fn local_port(&self) -> u16 {
+        self.local_port
+    }
+
+    pub(crate) fn remote_port(&self) -> u16 {
+        self.remote_port
+    }
+}
+
+impl Drop for LocalForward {
+    fn drop(&mut self) {
+        self.task.abort();
+    }
+}
+
+pub(crate) async fn open_local_forward(
+    raw_destination: &str,
+    local_port: u16,
+    remote_port: u16,
+) -> Result<LocalForward, SessionError> {
+    let profiles_path = crate::display::nebula_data_dir().join("ssh_profiles.json");
+    let raw = raw_destination.to_owned();
+    let (destination, profile) = tokio::task::spawn_blocking(move || {
+        let destination = SshDestination::resolve(&raw)?;
+        let profiles = crate::ssh_profiles::SshProfiles::load(&profiles_path)?;
+        Ok::<_, std::io::Error>((destination, profiles.for_destination(&raw)))
+    })
+    .await
+    .map_err(|error| format!("SSH 地址解析任务失败: {error}"))??;
+
+    let session = authenticated_session(&destination, &profile, None::<&NoopSshEventHost>).await?;
+    bind_forward(session, local_port, remote_port).await
+}
+
+async fn bind_forward(
+    session: SharedSession,
+    local_port: u16,
+    remote_port: u16,
+) -> Result<LocalForward, SessionError> {
+    let listener = TcpListener::bind((Ipv4Addr::LOCALHOST, local_port)).await?;
+    let local_port = listener.local_addr()?.port();
+    let task = tokio::spawn(async move {
+        let mut connections = JoinSet::new();
+        loop {
+            tokio::select! {
+                accepted = listener.accept(), if connections.len() < MAX_CONNECTIONS => {
+                    let Ok((mut local, peer)) = accepted else { break };
+                    let session = session.clone();
+                    connections.spawn(async move {
+                        let channel = super::lifecycle::network(
+                            "port-forward channel",
+                            session.channel_open_direct_tcpip(
+                                Ipv4Addr::LOCALHOST.to_string(),
+                                u32::from(remote_port),
+                                peer.ip().to_string(),
+                                u32::from(peer.port()),
+                            ),
+                        )
+                        .await?;
+                        let mut remote = channel.into_stream();
+                        tokio::io::copy_bidirectional(&mut local, &mut remote).await?;
+                        Ok::<(), SessionError>(())
+                    });
+                },
+                Some(result) = connections.join_next(), if !connections.is_empty() => {
+                    match result {
+                        Ok(Ok(())) => {},
+                        Ok(Err(error)) => log::warn!("SSH port-forward connection failed: {error}"),
+                        Err(error) => log::warn!("SSH port-forward task failed: {error}"),
+                    }
+                },
+            }
+        }
+    });
+
+    Ok(LocalForward { local_port, remote_port, task })
+}
+
+#[cfg(test)]
+mod tests;
diff --git a/nebula_app/src/ssh_session/forward/tests.rs b/nebula_app/src/ssh_session/forward/tests.rs
new file mode 100644
index 00000000..c2e2f484
--- /dev/null
+++ b/nebula_app/src/ssh_session/forward/tests.rs
@@ -0,0 +1,186 @@
+use super::*;
+use russh::keys::ssh_key::{Algorithm, PrivateKey};
+use russh::{client, server};
+use std::sync::Arc;
+use std::time::Duration;
+use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::net::TcpStream;
+use tokio::sync::mpsc;
+
+struct Echo {
+    opened: mpsc::UnboundedSender<u32>,
+    streams: JoinSet<()>,
+}
+
+impl server::Handler for Echo {
+    type Error = russh::Error;
+
+    async fn auth_none(&mut self, _: &str) -> Result<server::Auth, Self::Error> {
+        Ok(server::Auth::Accept)
+    }
+
+    async fn channel_open_direct_tcpip(
+        &mut self,
+        channel: russh::Channel<server::Msg>,
+        host: &str,
+        port: u32,
+        origin: &str,
+        _: u32,
+        reply: server::ChannelOpenHandle,
+        _: &mut server::Session,
+    ) -> Result<(), Self::Error> {
+        assert_eq!(host, "127.0.0.1");
+        assert_eq!(origin, "127.0.0.1");
+        if port == 1 {
+            return Ok(());
+        } // Exercise a real SSH channel-open rejection.
+        reply.accept().await;
+        self.opened.send(port).unwrap();
+        self.streams.spawn(async move {
+            let mut stream = channel.into_stream();
+            let mut bytes = [0; 128];
+            while let Ok(count) = stream.read(&mut bytes).await {
+                if count == 0 {
+                    break;
+                }
+                if stream.write_all(&bytes[..count]).await.is_err() {
+                    return;
+                }
+            }
+            let _ = stream.shutdown().await;
+        });
+        Ok(())
+    }
+}
+
+struct Fixture {
+    session: SharedSession,
+    opened: mpsc::UnboundedReceiver<u32>,
+    server: JoinHandle<()>,
+    _directory: tempfile::TempDir,
+}
+
+impl Drop for Fixture {
+    fn drop(&mut self) {
+        self.server.abort();
+    }
+}
+
+impl Fixture {
+    async fn new() -> Self {
+        let directory = tempfile::tempdir().unwrap();
+        let known_hosts = directory.path().join("known_hosts");
+        let key = PrivateKey::random(&mut rand::rng(), Algorithm::Ed25519).unwrap();
+        russh::keys::known_hosts::learn_known_hosts_path(
+            "forward.test",
+            22,
+            key.public_key(),
+            &known_hosts,
+        )
+        .unwrap();
+        let config = Arc::new(server::Config {
+            keys: vec![key],
+            auth_rejection_time: Duration::ZERO,
+            auth_rejection_time_initial: Some(Duration::ZERO),
+            ..Default::default()
+        });
+        let (client_io, server_io) = tokio::io::duplex(65536);
+        let (sender, opened) = mpsc::unbounded_channel();
+        let server = tokio::spawn(async move {
+            let session = server::run_stream(
+                config,
+                server_io,
+                Echo { opened: sender, streams: JoinSet::new() },
+            )
+            .await
+            .unwrap();
+            let _ = session.await;
+        });
+        let mut session = client::connect_stream(
+            Arc::new(client::Config::default()),
+            client_io,
+            super::super::ClientHandler {
+                host: "forward.test".into(),
+                port: 22,
+                allow_prompt: false,
+                handshake: super::super::lifecycle::Handshake::default(),
+                known_hosts_path: Some(known_hosts),
+            },
+        )
+        .await
+        .unwrap();
+        assert!(session.authenticate_none("fixture").await.unwrap().success());
+        Self { session: Arc::new(session), opened, server, _directory: directory }
+    }
+}
+
+fn check(future: impl std::future::Future<Output = ()>) {
+    tokio::runtime::Builder::new_current_thread().enable_all().build().unwrap().block_on(async {
+        tokio::time::timeout(Duration::from_secs(10), future)
+            .await
+            .expect("forward regression timed out");
+    });
+}
+
+#[test]
+fn forwards_bytes_and_half_close_and_releases_listener_and_connections() {
+    check(async {
+        let mut fixture = Fixture::new().await;
+        let forward = bind_forward(fixture.session.clone(), 0, 3000).await.unwrap();
+        let port = forward.local_port();
+        assert_eq!(forward.remote_port(), 3000);
+        assert!(bind_forward(fixture.session.clone(), port, 3001).await.is_err());
+        let mut socket = TcpStream::connect((Ipv4Addr::LOCALHOST, port)).await.unwrap();
+        assert_eq!(fixture.opened.recv().await, Some(3000));
+        socket.write_all(b"forwarded payload").await.unwrap();
+        socket.shutdown().await.unwrap();
+        let mut response = Vec::new();
+        socket.read_to_end(&mut response).await.unwrap();
+        assert_eq!(response, b"forwarded payload");
+        let mut active = TcpStream::connect((Ipv4Addr::LOCALHOST, port)).await.unwrap();
+        assert_eq!(fixture.opened.recv().await, Some(3000));
+        drop(forward);
+        let mut byte = [0];
+        assert!(matches!(active.read(&mut byte).await, Ok(0) | Err(_)));
+        TcpListener::bind((Ipv4Addr::LOCALHOST, port)).await.expect("listener released after drop");
+        assert!(
+            !fixture.session.is_closed(),
+            "stopping a forward preserves the shared SSH connection"
+        );
+    });
+}
+
+#[test]
+fn rejected_channel_closes_only_that_client() {
+    check(async {
+        let fixture = Fixture::new().await;
+        let forward = bind_forward(fixture.session.clone(), 0, 1).await.unwrap();
+        for _ in 0..2 {
+            let mut socket =
+                TcpStream::connect((Ipv4Addr::LOCALHOST, forward.local_port())).await.unwrap();
+            assert!(matches!(socket.read(&mut [0]).await, Ok(0) | Err(_)));
+        }
+        assert!(!fixture.session.is_closed());
+    });
+}
+
+#[test]
+fn channel_tasks_are_bounded_and_resume_after_a_client_closes() {
+    check(async {
+        let mut fixture = Fixture::new().await;
+        let forward = bind_forward(fixture.session.clone(), 0, 3000).await.unwrap();
+        let mut clients = Vec::new();
+        for _ in 0..MAX_CONNECTIONS {
+            clients.push(
+                TcpStream::connect((Ipv4Addr::LOCALHOST, forward.local_port())).await.unwrap(),
+            );
+            assert_eq!(fixture.opened.recv().await, Some(3000));
+        }
+        let _waiting =
+            TcpStream::connect((Ipv4Addr::LOCALHOST, forward.local_port())).await.unwrap();
+        tokio::task::yield_now().await;
+        assert!(fixture.opened.try_recv().is_err());
+        clients.pop().unwrap().shutdown().await.unwrap();
+        assert_eq!(fixture.opened.recv().await, Some(3000));
+    });
+}