fix(http): stream git smart HTTP responses

Return channel-backed response bodies for receive-pack and upload-pack so Git stdout reaches clients as soon as it is produced instead of after the subprocess exits.

This fixes a production issue seen when pushing large repositories with ngit-cli: large receive-pack pushes could sit silent while Git resolved deltas and checked connectivity, causing libgit2 clients to hit their per-recv timeout instead of observing sideband progress.

The streaming task now owns subprocess cleanup, stderr draining, lifecycle locks, metrics, and post-push purgatory promotion. Add dedicated /prs/ regression coverage to prove receive-pack stdout streams before EOF while zero-ref /prs/ cleanup waits until the in-flight push finishes.

Fixes nostr:nevent1qqsyjly2u925qdc59cmxlxarvwagnr7pd6gpnr776x2mnfz6mdhn33cpz3mhxue69uhhyetvv9ujumn8d96zuer9wcj596eq

src/git/handlers.rs

@@ -2,25 +2,33 @@
 //!
 //! This module implements the HTTP handlers for Git Smart HTTP protocol.
 
-use http_body_util::Full;
-use hyper::{body::Bytes, Response, StatusCode};
+use futures_util::stream;
+use http_body_util::{BodyExt, StreamBody};
+use hyper::{body::Bytes, body::Frame, Response, StatusCode};
 use nostr_relay_builder::LocalRelay;
 use nostr_sdk::prelude::*;
 use std::collections::HashSet;
+use std::io;
 use std::path::PathBuf;
 use std::sync::Arc;
 use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::sync::mpsc;
 use tracing::{debug, error, info, warn};
 
 use super::protocol::{GitService, PktLine};
 use super::subprocess::GitSubprocess;
+use super::{full_body, GitResponseBody};
 
 use crate::git::authorization::{authorize_push, parse_pushed_refs};
 use crate::git::sync::{process_newly_available_git_data, PurgatoryPromotionHooks};
+use crate::metrics::Metrics;
 use crate::nostr::lifecycle::LifecycleReadGuard;
 use crate::nostr::SharedDatabase;
 use crate::purgatory::Purgatory;
 
+pub(crate) const STREAM_CHANNEL_DEPTH: usize = 8;
+const STREAM_CHUNK_SIZE: usize = 8 * 1024;
+
 /// Handle GET /info/refs?service=git-{upload,receive}-pack
 ///
 /// This advertises the repository's refs to the client.
@@ -28,7 +36,7 @@ pub async fn handle_info_refs(
     repo_path: PathBuf,
     service: GitService,
     git_protocol: Option<&str>,
-) -> Result<Response<Full<Bytes>>, GitError> {
+) -> Result<Response<GitResponseBody>, GitError> {
     debug!(
         "Handling info/refs for {:?} with service {:?}",
         repo_path, service
@@ -96,7 +104,7 @@ pub async fn handle_info_refs(
         .status(StatusCode::OK)
         .header("content-type", service.advertisement_content_type())
         .header("cache-control", "no-cache")
-        .body(Full::new(Bytes::from(response_body)))
+        .body(full_body(response_body))
         .unwrap())
 }
 
@@ -178,7 +186,22 @@ pub(crate) fn build_git_protocol_error_response(
     service: GitService,
     error_message: &str,
     request_body: Option<&[u8]>,
-) -> Response<Full<Bytes>> {
+) -> Response<GitResponseBody> {
+    let err_pktline = encode_err_pktline(service, error_message, request_body);
+
+    Response::builder()
+        .status(StatusCode::OK)
+        .header("content-type", service.result_content_type())
+        .header("cache-control", "no-cache")
+        .body(full_body(err_pktline))
+        .unwrap()
+}
+
+fn encode_err_pktline(
+    service: GitService,
+    error_message: &str,
+    request_body: Option<&[u8]>,
+) -> Vec<u8> {
     // Format: "ERR <message>\n"
     let err_content = format!("ERR {}\n", error_message.trim());
 
@@ -189,21 +212,26 @@ pub(crate) fn build_git_protocol_error_response(
             .map(client_negotiated_sideband_64k)
             .unwrap_or(false);
 
-    let err_pktline = if use_sideband {
+    if use_sideband {
         let mut framed = Vec::with_capacity(1 + err_content.len());
         framed.push(0x03); // band 3 = error
         framed.extend_from_slice(err_content.as_bytes());
         PktLine::data(framed).encode()
     } else {
         PktLine::data(err_content.as_bytes()).encode()
-    };
+    }
+}
 
-    Response::builder()
-        .status(StatusCode::OK)
-        .header("content-type", service.result_content_type())
-        .header("cache-control", "no-cache")
-        .body(Full::new(Bytes::from(err_pktline)))
-        .unwrap()
+pub(crate) fn err_pktline_frame(
+    service: GitService,
+    error_message: &str,
+    request_body: Option<&[u8]>,
+) -> Frame<Bytes> {
+    Frame::data(Bytes::from(encode_err_pktline(
+        service,
+        error_message,
+        request_body,
+    )))
 }
 
 /// Check if a git process failure is a protocol error (vs transport error).
@@ -222,12 +250,105 @@ pub(crate) fn is_git_protocol_error(exit_code: Option<i32>, stderr: &[u8]) -> bo
     exit_code == Some(128) && !stderr.is_empty()
 }
 
+/// Build the common channel-backed Git response used by streaming handlers.
+///
+/// The handler returns this response before the Git child has necessarily
+/// exited. A background task sends `Frame<Bytes>` values into `rx`; dropping the
+/// sender closes the HTTP body. Kept `pub(crate)` so `/prs/` receive-pack can
+/// use the same wire shape as the standard endpoints.
+pub(crate) fn streaming_response(
+    service: GitService,
+    rx: mpsc::Receiver<Result<Frame<Bytes>, io::Error>>,
+) -> Response<GitResponseBody> {
+    let body_stream = stream::unfold(rx, |mut rx| async {
+        rx.recv().await.map(|item| (item, rx))
+    });
+    let body = BodyExt::boxed(StreamBody::new(body_stream));
+
+    Response::builder()
+        .status(StatusCode::OK)
+        .header("content-type", service.result_content_type())
+        .header("cache-control", "no-cache")
+        .body(body)
+        .unwrap()
+}
+
+/// Record Git metrics from either the request handler or the detached streaming
+/// task without repeating the `Option<Metrics>` plumbing at every callsite.
+pub(crate) fn record_git_operation(metrics: &Option<Arc<Metrics>>, operation: &str, status: &str) {
+    if let Some(metrics) = metrics {
+        metrics.record_git_operation(operation, status);
+    }
+}
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub(crate) enum PumpResult {
+    /// Stdout reached EOF. `sent_stdout` tells callers whether it is still safe
+    /// to synthesize a Git `ERR` pkt-line for a late process failure.
+    Eof { sent_stdout: bool },
+    /// The response receiver was dropped, usually because the HTTP client
+    /// disconnected. Callers should stop the child process and clean up state.
+    ClientDisconnected,
+    /// Reading stdout failed. The read error has already been sent through the
+    /// body channel so Hyper can terminate the response stream.
+    ReadError,
+}
+
+/// Forward Git stdout into the streaming response channel.
+///
+/// This is intentionally only the stdout pump. The caller still owns process
+/// termination, stderr collection, protocol-error classification, and any
+/// endpoint-specific cleanup after the stream ends.
+pub(crate) async fn pump_stdout_to_channel<R>(
+    mut stdout: R,
+    tx: &mpsc::Sender<Result<Frame<Bytes>, io::Error>>,
+) -> PumpResult
+where
+    R: tokio::io::AsyncRead + Unpin,
+{
+    let mut read_buf = [0_u8; STREAM_CHUNK_SIZE];
+    let mut sent_stdout = false;
+
+    loop {
+        match stdout.read(&mut read_buf).await {
+            Ok(0) => return PumpResult::Eof { sent_stdout },
+            Ok(n) => {
+                sent_stdout = true;
+                if send_body_bytes(tx, read_buf[..n].to_vec()).await.is_err() {
+                    return PumpResult::ClientDisconnected;
+                }
+            }
+            Err(e) => {
+                let _ = tx.send(Err(e)).await;
+                return PumpResult::ReadError;
+            }
+        }
+    }
+}
+
+/// Drain Git stderr for later logging or Git protocol error synthesis.
+///
+/// Streaming handlers run this concurrently with stdout pumping so the child
+/// cannot block on a full stderr pipe while the HTTP body is still being read.
+pub(crate) async fn read_stderr_to_end<R>(mut stderr: R) -> Vec<u8>
+where
+    R: tokio::io::AsyncRead + Unpin,
+{
+    let mut stderr_output = Vec::new();
+    if let Err(e) = stderr.read_to_end(&mut stderr_output).await {
+        warn!("Failed to read git subprocess stderr: {}", e);
+    }
+    stderr_output
+}
+
 /// Handle POST /git-upload-pack (clone/fetch)
 pub async fn handle_upload_pack(
     repo_path: PathBuf,
     request_body: Bytes,
     git_protocol: Option<&str>,
-) -> Result<Response<Full<Bytes>>, GitError> {
+    repo_lifecycle_guard: Option<LifecycleReadGuard>,
+    metrics: Option<Arc<Metrics>>,
+) -> Result<Response<GitResponseBody>, GitError> {
     debug!("Handling upload-pack for {:?}", repo_path);
 
     if !repo_path.exists() {
@@ -248,60 +369,101 @@ pub async fn handle_upload_pack(
         drop(stdin);
     }
 
-    // Read response from git's stdout
-    let mut output = Vec::new();
-    let mut stderr_output = Vec::new();
-
-    if let Some(stdout) = git.take_stdout() {
-        let mut stdout = stdout;
-        stdout.read_to_end(&mut output).await.map_err(|e| {
-            error!("Failed to read git upload-pack stdout: {}", e);
-            GitError::IoError(e)
-        })?;
-    }
-
-    if let Some(stderr) = git.take_stderr() {
-        let mut stderr = stderr;
-        stderr.read_to_end(&mut stderr_output).await.map_err(|e| {
-            error!("Failed to read git upload-pack stderr: {}", e);
-            GitError::IoError(e)
-        })?;
-    }
-
-    // Wait for process
-    let status = git.wait().await.map_err(|e| {
-        error!("Failed to wait for git upload-pack process: {}", e);
-        GitError::IoError(e)
+    let stdout = git.take_stdout().ok_or_else(|| {
+        GitError::IoError(io::Error::new(
+            io::ErrorKind::BrokenPipe,
+            "git upload-pack stdout unavailable",
+        ))
     })?;
+    let stderr = git.take_stderr();
 
-    if !status.success() {
+    // Stream upload-pack stdout as Git produces it instead of buffering the
+    // whole fetch response in memory. The detached task below owns the child
+    // until EOF so the request future can return the HTTP body immediately.
+    let (tx, rx) = mpsc::channel::<Result<Frame<Bytes>, io::Error>>(STREAM_CHANNEL_DEPTH);
+
+    tokio::spawn(async move {
+        stream_upload_pack_output(git, stdout, stderr, tx, repo_lifecycle_guard, metrics).await;
+    });
+
+    Ok(streaming_response(GitService::UploadPack, rx))
+}
+
+async fn stream_upload_pack_output<S, E>(
+    mut git: GitSubprocess,
+    stdout: S,
+    stderr: Option<E>,
+    tx: mpsc::Sender<Result<Frame<Bytes>, io::Error>>,
+    repo_lifecycle_guard: Option<LifecycleReadGuard>,
+    metrics: Option<Arc<Metrics>>,
+) where
+    S: tokio::io::AsyncRead + Unpin + Send + 'static,
+    E: tokio::io::AsyncRead + Unpin + Send + 'static,
+{
+    let stderr_task = stderr.map(|stderr| tokio::spawn(read_stderr_to_end(stderr)));
+    let pump_result = pump_stdout_to_channel(stdout, &tx).await;
+
+    if !matches!(pump_result, PumpResult::Eof { .. }) {
+        let _ = git.kill().await;
+    }
+
+    let status = match git.wait().await {
+        Ok(status) => status,
+        Err(e) => {
+            let _ = tx.send(Err(e)).await;
+            drop(repo_lifecycle_guard);
+            record_git_operation(&metrics, "clone", "error");
+            return;
+        }
+    };
+
+    // Keep the repository lifecycle read lock until git-upload-pack exits so
+    // deletion/archive/restore cannot remove or replace the bare repository
+    // while Git is still reading it in the detached streaming task.
+    drop(repo_lifecycle_guard);
+
+    let stderr_output = match stderr_task {
+        Some(task) => task.await.unwrap_or_default(),
+        None => Vec::new(),
+    };
+
+    if !status.success() && matches!(pump_result, PumpResult::Eof { sent_stdout: false }) {
+        record_git_operation(&metrics, "clone", "error");
         let stderr_str = String::from_utf8_lossy(&stderr_output);
+        let msg = if stderr_str.trim().is_empty() {
+            format!("git upload-pack failed with code {:?}", status.code())
+        } else {
+            stderr_str.to_string()
+        };
 
-        // Check if this is a git protocol error (exit code 128 with stderr)
-        // Protocol errors should be returned as HTTP 200 with ERR pkt-line
         if is_git_protocol_error(status.code(), &stderr_output) {
             warn!(
                 "Git upload-pack protocol error (returning ERR pkt-line): {}",
                 stderr_str
             );
-            return Ok(build_git_protocol_error_response(
-                GitService::UploadPack,
-                &stderr_str,
-                None,
-            ));
+        } else {
+            error!("Git upload-pack failed: {}", stderr_str);
         }
 
-        // Transport errors (spawn failures, signals, etc.) remain as HTTP 500
-        error!("Git upload-pack failed: {}", stderr_str);
-        return Err(GitError::GitFailed(status.code()));
+        // The streaming response headers have already been sent. If Git failed
+        // before producing any stdout, surface a protocol-visible ERR pkt-line
+        // instead of silently completing an empty HTTP 200 body.
+        let _ = tx
+            .send(Ok(err_pktline_frame(GitService::UploadPack, &msg, None)))
+            .await;
+    } else if !status.success() {
+        record_git_operation(&metrics, "clone", "error");
+        let stderr_str = String::from_utf8_lossy(&stderr_output);
+        error!(
+            "Git upload-pack failed after streaming stdout: {}",
+            stderr_str
+        );
+    } else if matches!(pump_result, PumpResult::Eof { .. }) {
+        debug!("Git upload-pack stream completed successfully");
+        record_git_operation(&metrics, "clone", "success");
+    } else {
+        record_git_operation(&metrics, "clone", "error");
     }
-
-    Ok(Response::builder()
-        .status(StatusCode::OK)
-        .header("content-type", GitService::UploadPack.result_content_type())
-        .header("cache-control", "no-cache")
-        .body(Full::new(Bytes::from(output)))
-        .unwrap())
 }
 
 /// Handle POST /git-receive-pack (push)
@@ -338,8 +500,9 @@ pub async fn handle_receive_pack(
     git_data_path: &str,
     git_protocol: Option<&str>,
     repo_lifecycle_guard: Option<LifecycleReadGuard>,
-    promotion_hooks: Option<&dyn PurgatoryPromotionHooks>,
-) -> Result<Response<Full<Bytes>>, GitError> {
+    promotion_hooks: Option<Arc<dyn PurgatoryPromotionHooks>>,
+    metrics: Option<Arc<Metrics>>,
+) -> Result<Response<GitResponseBody>, GitError> {
     debug!("Handling receive-pack for {:?}", repo_path);
 
     if !repo_path.exists() {
@@ -366,6 +529,7 @@ pub async fn handle_receive_pack(
         Ok(auth_result) => {
             if !auth_result.authorized {
                 warn!("Push rejected for {}: {}", identifier, auth_result.reason);
+                record_git_operation(&metrics, "push", "error");
                 return Ok(build_git_protocol_error_response(
                     GitService::ReceivePack,
                     &format!("authorisation failed: {}", auth_result.reason),
@@ -383,6 +547,7 @@ pub async fn handle_receive_pack(
         }
         Err(e) => {
             warn!("Authorization check failed for {}: {}", identifier, e);
+            record_git_operation(&metrics, "push", "error");
             return Ok(build_git_protocol_error_response(
                 GitService::ReceivePack,
                 &format!("authorisation failed: {}", e),
@@ -404,55 +569,147 @@ pub async fn handle_receive_pack(
         drop(stdin);
     }
 
-    // Read response from git's stdout
-    let mut output = Vec::new();
-    let mut stderr_output = Vec::new();
-
-    if let Some(stdout) = git.take_stdout() {
-        let mut stdout = stdout;
-        stdout.read_to_end(&mut output).await.map_err(|e| {
-            error!("Failed to read git receive-pack stdout: {}", e);
-            GitError::IoError(e)
-        })?;
-    }
-
-    if let Some(stderr) = git.take_stderr() {
-        let mut stderr = stderr;
-        stderr.read_to_end(&mut stderr_output).await.map_err(|e| {
-            error!("Failed to read git receive-pack stderr: {}", e);
-            GitError::IoError(e)
-        })?;
-    }
-
-    // Wait for process
-    let status = git.wait().await.map_err(|e| {
-        error!("Failed to wait for git receive-pack process: {}", e);
-        GitError::IoError(e)
+    let stdout = git.take_stdout().ok_or_else(|| {
+        GitError::IoError(io::Error::new(
+            io::ErrorKind::BrokenPipe,
+            "git receive-pack stdout unavailable",
+        ))
     })?;
+    let stderr = git.take_stderr();
+
+    // Do not buffer receive-pack stdout: for large pushes Git may spend tens of
+    // seconds resolving deltas/checking connectivity after the client finishes
+    // uploading. Streaming sideband progress keeps libgit2 clients from
+    // hitting their per-recv timeout during that otherwise-silent window.
+    let (tx, rx) = mpsc::channel::<Result<Frame<Bytes>, io::Error>>(STREAM_CHANNEL_DEPTH);
+    let pushed_refs = parse_pushed_refs(&request_body);
+    let new_oids: HashSet<String> = pushed_refs
+        .iter()
+        .filter(|(_, new_oid, _)| new_oid != "0000000000000000000000000000000000000000")
+        .map(|(_, new_oid, _)| new_oid.clone())
+        .collect();
+    let request_body_for_errors = request_body.clone();
+    let identifier = identifier.to_owned();
+    let git_data_path = git_data_path.to_owned();
+
+    tokio::spawn(async move {
+        stream_receive_pack_output(
+            git,
+            stdout,
+            stderr,
+            tx,
+            repo_path,
+            new_oids,
+            database,
+            relay,
+            identifier,
+            purgatory,
+            git_data_path,
+            request_body_for_errors,
+            repo_lifecycle_guard,
+            promotion_hooks,
+            metrics,
+        )
+        .await;
+    });
+
+    Ok(streaming_response(GitService::ReceivePack, rx))
+}
+
+#[allow(clippy::too_many_arguments)]
+async fn stream_receive_pack_output<S, E>(
+    mut git: GitSubprocess,
+    stdout: S,
+    stderr: Option<E>,
+    tx: mpsc::Sender<Result<Frame<Bytes>, io::Error>>,
+    repo_path: PathBuf,
+    new_oids: HashSet<String>,
+    database: SharedDatabase,
+    relay: LocalRelay,
+    identifier: String,
+    purgatory: Arc<Purgatory>,
+    git_data_path: String,
+    request_body: Bytes,
+    repo_lifecycle_guard: Option<LifecycleReadGuard>,
+    promotion_hooks: Option<Arc<dyn PurgatoryPromotionHooks>>,
+    metrics: Option<Arc<Metrics>>,
+) where
+    S: tokio::io::AsyncRead + Unpin + Send + 'static,
+    E: tokio::io::AsyncRead + Unpin + Send + 'static,
+{
+    let stderr_task = stderr.map(|stderr| tokio::spawn(read_stderr_to_end(stderr)));
+    let pump_result = pump_stdout_to_channel(stdout, &tx).await;
+
+    if !matches!(pump_result, PumpResult::Eof { .. }) {
+        let _ = git.kill().await;
+    }
+
+    let status = match git.wait().await {
+        Ok(status) => status,
+        Err(e) => {
+            let _ = tx.send(Err(e)).await;
+            drop(repo_lifecycle_guard);
+            record_git_operation(&metrics, "push", "error");
+            return;
+        }
+    };
+
+    let stderr_output = match stderr_task {
+        Some(task) => task.await.unwrap_or_default(),
+        None => Vec::new(),
+    };
+
+    let sent_stdout = match pump_result {
+        PumpResult::Eof { sent_stdout } => sent_stdout,
+        PumpResult::ClientDisconnected | PumpResult::ReadError => {
+            drop(repo_lifecycle_guard);
+            record_git_operation(&metrics, "push", "error");
+            return;
+        }
+    };
 
     if !status.success() {
         drop(repo_lifecycle_guard);
+        record_git_operation(&metrics, "push", "error");
         let stderr_str = String::from_utf8_lossy(&stderr_output);
 
-        // Check if this is a git protocol error (exit code 128 with stderr)
-        // Protocol errors should be returned as HTTP 200 with ERR pkt-line
         if is_git_protocol_error(status.code(), &stderr_output) {
             warn!(
                 "Git receive-pack protocol error (returning ERR pkt-line): {}",
                 stderr_str
             );
-            return Ok(build_git_protocol_error_response(
-                GitService::ReceivePack,
-                &stderr_str,
-                Some(&request_body),
-            ));
+            if !sent_stdout {
+                let _ = tx
+                    .send(Ok(err_pktline_frame(
+                        GitService::ReceivePack,
+                        &stderr_str,
+                        Some(&request_body),
+                    )))
+                    .await;
+            }
+        } else {
+            error!("Git receive-pack failed: {}", stderr_str);
+            if !sent_stdout {
+                let msg = if stderr_str.trim().is_empty() {
+                    format!("git receive-pack failed with code {:?}", status.code())
+                } else {
+                    stderr_str.to_string()
+                };
+                let _ = tx
+                    .send(Ok(err_pktline_frame(
+                        GitService::ReceivePack,
+                        &msg,
+                        Some(&request_body),
+                    )))
+                    .await;
+            }
         }
-
-        // Transport errors (spawn failures, signals, etc.) remain as HTTP 500
-        error!("Git receive-pack failed: {}", stderr_str);
-        return Err(GitError::GitFailed(status.code()));
+        return;
     }
 
+    debug!("Git receive-pack stream completed successfully");
+    record_git_operation(&metrics, "push", "success");
+
     // Release the repository lifecycle read lock once git-receive-pack itself
     // has finished. The lock's purpose is to keep deletion/archive/restore from
     // removing or replacing the bare repository while Git is actively reading or
@@ -462,33 +719,19 @@ pub async fn handle_receive_pack(
     // the subprocess boundary would deadlock.
     drop(repo_lifecycle_guard);
 
-    // Process newly available git data using the unified function
-    // This handles:
-    // - Discovering satisfiable events from purgatory (state events and PR events)
-    // - Syncing OIDs to authorized owner repos
-    // - Aligning refs (+ setting HEAD) in all owner repos
-    // - Saving events to database
-    // - Notifying WebSocket subscribers
-    // - Removing from purgatory
-    //
-    // Parse pushed refs to collect new OIDs
-    let pushed_refs = parse_pushed_refs(&request_body);
-    let new_oids: HashSet<String> = pushed_refs
-        .iter()
-        .filter(|(_, new_oid, _)| new_oid != "0000000000000000000000000000000000000000")
-        .map(|(_, new_oid, _)| new_oid.clone())
-        .collect();
-
-    let git_data_path_buf = std::path::Path::new(git_data_path);
-
+    // Git's receive-pack response has already been streamed verbatim, including
+    // its final flush. Run GRASP follow-up work before closing the HTTP body so
+    // push completion still implies local refs/HEAD are aligned, but treat those
+    // follow-up failures as internal/log-only rather than client-visible push
+    // rejections.
     match process_newly_available_git_data(
         &repo_path,
         &new_oids,
         &database,
         Some(&relay),
         &purgatory,
-        git_data_path_buf,
-        promotion_hooks,
+        std::path::Path::new(&git_data_path),
+        promotion_hooks.as_deref(),
     )
     .await
     {
@@ -516,16 +759,13 @@ pub async fn handle_receive_pack(
             );
         }
     }
+}
 
-    Ok(Response::builder()
-        .status(StatusCode::OK)
-        .header(
-            "content-type",
-            GitService::ReceivePack.result_content_type(),
-        )
-        .header("cache-control", "no-cache")
-        .body(Full::new(Bytes::from(output)))
-        .unwrap())
+async fn send_body_bytes(
+    tx: &mpsc::Sender<Result<Frame<Bytes>, io::Error>>,
+    bytes: Vec<u8>,
+) -> Result<(), mpsc::error::SendError<Result<Frame<Bytes>, io::Error>>> {
+    tx.send(Ok(Frame::data(Bytes::from(bytes)))).await
 }
 
 /// Errors that can occur in Git handlers
@@ -591,7 +831,7 @@ mod tests {
         out
     }
 
-    async fn response_body_bytes(resp: Response<Full<Bytes>>) -> Vec<u8> {
+    async fn response_body_bytes(resp: Response<GitResponseBody>) -> Vec<u8> {
         resp.into_body()
             .collect()
             .await
@@ -698,6 +938,39 @@ mod tests {
         assert_eq!(raw[4], b'E');
     }
 
+    #[tokio::test]
+    async fn streaming_body_yields_stdout_before_eof() {
+        use tokio::io::AsyncWriteExt;
+        use tokio::time::{timeout, Duration};
+
+        let (mut writer, reader) = tokio::io::duplex(64);
+        let (tx, rx) = mpsc::channel::<Result<Frame<Bytes>, io::Error>>(STREAM_CHANNEL_DEPTH);
+
+        let pump_task = tokio::spawn(async move { pump_stdout_to_channel(reader, &tx).await });
+        writer.write_all(b"progress chunk").await.unwrap();
+
+        let mut body = streaming_response(GitService::ReceivePack, rx).into_body();
+        let frame = timeout(Duration::from_secs(1), body.frame())
+            .await
+            .expect("stream should yield first stdout chunk before EOF")
+            .expect("stream should still be open")
+            .expect("stdout chunk should not be a body error");
+
+        let data = frame.into_data().expect("frame should contain data");
+        assert_eq!(data, Bytes::from_static(b"progress chunk"));
+
+        assert!(
+            !pump_task.is_finished(),
+            "stdout pump should still be waiting for EOF after yielding first chunk"
+        );
+
+        drop(writer);
+        assert_eq!(
+            pump_task.await.unwrap(),
+            PumpResult::Eof { sent_stdout: true }
+        );
+    }
+
     #[tokio::test]
     async fn receive_pack_err_without_body_is_not_wrapped() {
         // No request body → conservative path: do not wrap.

src/git/mod.rs

@@ -24,10 +24,36 @@ pub mod protocol;
 pub mod subprocess;
 pub mod sync;
 
+use std::convert::Infallible;
 use std::path::{Path, PathBuf};
 use std::process::Command;
+
+use http_body_util::combinators::BoxBody;
+use http_body_util::{BodyExt, Full};
+use hyper::body::Bytes;
 use tracing::{debug, info};
 
+/// Unified HTTP response body for every route served by ngit-grasp.
+///
+/// This boxed body can hold either a buffered [`Full<Bytes>`] payload (for
+/// small responses such as advertisements, protocol-error pkt-lines, JSON
+/// pages, and static assets) or a streaming body backed by a channel (for live
+/// `git-receive-pack` / `git-upload-pack` stdout). Using one body type keeps
+/// the Hyper service response type unified.
+pub type GitResponseBody = BoxBody<Bytes, std::io::Error>;
+
+/// Wrap an in-memory payload as a one-shot [`GitResponseBody`].
+pub fn full_body(bytes: impl Into<Bytes>) -> GitResponseBody {
+    Full::new(bytes.into())
+        .map_err(|never: Infallible| match never {})
+        .boxed()
+}
+
+/// Explicit empty response body.
+pub fn empty_body() -> GitResponseBody {
+    full_body(Bytes::new())
+}
+
 /// Parse a Git repository path from URL components
 ///
 /// Converts /<npub>/<identifier>.git/* to a filesystem path

src/grasp06/fetch.rs

@@ -12,18 +12,20 @@
 //!
 //! Receive-pack lives in [`crate::grasp06::receive`].
 
-use http_body_util::Full;
 use hyper::body::Bytes;
 use hyper::Response;
 use std::path::Path;
 use std::process::Command;
+use std::sync::Arc;
 use tempfile::TempDir;
 use tracing::{debug, warn};
 
 use crate::git::handlers::{handle_info_refs, handle_upload_pack, GitError};
 use crate::git::protocol::GitService;
+use crate::git::GitResponseBody;
 use crate::grasp06::endpoint::PrsUrl;
 use crate::grasp06::paths::prs_repo_path;
+use crate::metrics::Metrics;
 
 /// Handle `GET /prs/<npub>/<id>.git/info/refs?service=...`.
 ///
@@ -38,7 +40,7 @@ pub async fn handle_prs_info_refs(
     git_data_path: &str,
     service: GitService,
     git_protocol: Option<&str>,
-) -> Result<Response<Full<Bytes>>, GitError> {
+) -> Result<Response<GitResponseBody>, GitError> {
     let real_repo = prs_repo_path(
         Path::new(git_data_path),
         &prs.submitter.to_hex(),
@@ -75,7 +77,8 @@ pub async fn handle_prs_upload_pack(
     git_data_path: &str,
     body: Bytes,
     git_protocol: Option<&str>,
-) -> Result<Response<Full<Bytes>>, GitError> {
+    metrics: Option<Arc<Metrics>>,
+) -> Result<Response<GitResponseBody>, GitError> {
     let real_repo = prs_repo_path(
         Path::new(git_data_path),
         &prs.submitter.to_hex(),
@@ -87,7 +90,7 @@ pub async fn handle_prs_upload_pack(
             "/prs/ upload-pack: real repo found at {} — delegating",
             real_repo.display()
         );
-        return handle_upload_pack(real_repo, body, git_protocol).await;
+        return handle_upload_pack(real_repo, body, git_protocol, None, metrics).await;
     }
 
     debug!(
@@ -96,10 +99,74 @@ pub async fn handle_prs_upload_pack(
         prs.identifier
     );
     let temp = init_empty_bare_repo()?;
+    handle_prs_upload_pack_buffered(temp, body, git_protocol).await
+}
+
+async fn handle_prs_upload_pack_buffered(
+    temp: TempDir,
+    request_body: Bytes,
+    git_protocol: Option<&str>,
+) -> Result<Response<GitResponseBody>, GitError> {
+    use crate::git::full_body;
+    use crate::git::handlers::{build_git_protocol_error_response, is_git_protocol_error};
+    use crate::git::subprocess::GitSubprocess;
+    use hyper::StatusCode;
+    use tokio::io::{AsyncReadExt, AsyncWriteExt};
+
     let repo_path = temp.path().to_path_buf();
-    let response = handle_upload_pack(repo_path, body, git_protocol).await;
+    let mut git = GitSubprocess::spawn(GitService::UploadPack, &repo_path, false, git_protocol)
+        .map_err(GitError::ProcessSpawnFailed)?;
+
+    if let Some(mut stdin) = git.take_stdin() {
+        stdin
+            .write_all(&request_body)
+            .await
+            .map_err(GitError::IoError)?;
+        drop(stdin);
+    }
+
+    let mut output = Vec::new();
+    let mut stderr_output = Vec::new();
+
+    if let Some(mut stdout) = git.take_stdout() {
+        stdout
+            .read_to_end(&mut output)
+            .await
+            .map_err(GitError::IoError)?;
+    }
+
+    if let Some(mut stderr) = git.take_stderr() {
+        stderr
+            .read_to_end(&mut stderr_output)
+            .await
+            .map_err(GitError::IoError)?;
+    }
+
+    let status = git.wait().await.map_err(GitError::IoError)?;
+
+    // Keep the TempDir alive until git has fully exited. The standard
+    // upload-pack handler streams in a detached task; using that for this
+    // synthesized repo would race with dropping `temp` and deleting the repo.
     drop(temp);
-    response
+
+    if !status.success() {
+        let stderr_str = String::from_utf8_lossy(&stderr_output);
+        if is_git_protocol_error(status.code(), &stderr_output) {
+            return Ok(build_git_protocol_error_response(
+                GitService::UploadPack,
+                &stderr_str,
+                None,
+            ));
+        }
+        return Err(GitError::GitFailed(status.code()));
+    }
+
+    Ok(Response::builder()
+        .status(StatusCode::OK)
+        .header("content-type", GitService::UploadPack.result_content_type())
+        .header("cache-control", "no-cache")
+        .body(full_body(output))
+        .unwrap())
 }
 
 /// Create a fresh empty bare repo in a temp directory.

src/grasp06/receive.rs

@@ -27,50 +27,65 @@
 //!    own ref locking handles intra-push concurrency, and the
 //!    `in_flight` counter is what off-push cleanup paths consult to know
 //!    a push is active.
-//! 4. Runs `git-receive-pack` against the repo, mirroring the subprocess
-//!    plumbing in [`crate::git::handlers::handle_receive_pack`].
-//! 5. For each accepted `refs/nostr/<event-id>` ref, re-runs the shared
-//!    pre-validation as a **race safety net** — an event for one of the
-//!    pushed ids may have arrived via WebSocket during the receive-pack
-//!    window. On mismatch the ref is deleted (and any populated purgatory
-//!    entry is dropped). When neither the DB nor purgatory knows about
-//!    the event a scoped PR placeholder is added so the standard
-//!    30-minute purgatory sweep can clean it up if the event never
+//! 4. Starts `git-receive-pack`, writes the full request body to the child, and
+//!    immediately returns an HTTP response whose body is backed by a bounded
+//!    channel. From this point on Hyper can stream stdout to the client while a
+//!    detached task owns the subprocess, stderr, and all post-push work.
+//! 5. In the detached task, each stdout chunk is forwarded as it is read. If Git
+//!    exits with a protocol-level error before writing stdout, the task sends a
+//!    Git `ERR` pkt-line through the same stream so clients still see a normal
+//!    receive-pack failure. If stdout has already been sent, the task cannot
+//!    safely append a late protocol error without corrupting the Git stream, so
+//!    it logs and performs cleanup instead.
+//! 6. After a successful receive-pack, for each accepted
+//!    `refs/nostr/<event-id>` ref, re-runs the shared pre-validation as a
+//!    **race safety net** — an event for one of the pushed ids may have arrived
+//!    via WebSocket during the receive-pack window. On mismatch the ref is
+//!    deleted (and any populated purgatory entry is dropped). When neither the
+//!    DB nor purgatory knows about the event a scoped PR placeholder is added so
+//!    the standard 30-minute purgatory sweep can clean it up if the event never
 //!    arrives.
-//! 6. Re-acquires the per-path mutex briefly, decrements `in_flight`,
-//!    and — if no other push is in flight and the repo has zero refs
-//!    left — removes the bare directory. Always runs, including on
-//!    receive-pack protocol errors, so a failed push that has just
-//!    initialised an empty repo does not leak it.
-//! 7. Triggers the standard purgatory-release path via
-//!    [`crate::git::sync::process_newly_available_git_data`] so PR events
-//!    already in purgatory waiting for these commits get promoted.
+//! 7. The detached task re-acquires the per-path mutex briefly, decrements
+//!    `in_flight`, and — if no other push is in flight and the repo has zero
+//!    refs left — removes the bare directory. This runs on success, protocol
+//!    errors, client disconnects, and subprocess I/O failures, so a failed push
+//!    that has just initialised an empty repo does not leak it.
+//! 8. Finally, on successful pushes where the repo still exists, the detached
+//!    task triggers the standard purgatory-release path via
+//!    [`crate::git::sync::process_newly_available_git_data`] so PR events already
+//!    in purgatory waiting for these commits get promoted.
 
 use std::collections::HashSet;
+use std::io;
 use std::path::{Path, PathBuf};
 use std::sync::atomic::{AtomicUsize, Ordering};
 use std::sync::Arc;
 
 use dashmap::DashMap;
-use http_body_util::Full;
-use hyper::body::Bytes;
-use hyper::{Response, StatusCode};
+use hyper::body::{Bytes, Frame};
+use hyper::Response;
 use nostr_relay_builder::LocalRelay;
 use nostr_sdk::prelude::*;
 use std::sync::Mutex;
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::io::AsyncWriteExt;
+use tokio::sync::mpsc;
 use tracing::{debug, error, info, warn};
 
 use crate::git::authorization::{
     parse_pushed_refs, pre_validate_refs_nostr_push, NostrRefPreValidation, PrsUrlConstraints,
 };
-use crate::git::handlers::{build_git_protocol_error_response, is_git_protocol_error, GitError};
+use crate::git::handlers::{
+    build_git_protocol_error_response, err_pktline_frame, is_git_protocol_error,
+    pump_stdout_to_channel, read_stderr_to_end, record_git_operation, streaming_response, GitError,
+    PumpResult, STREAM_CHANNEL_DEPTH,
+};
 use crate::git::protocol::GitService;
 use crate::git::subprocess::GitSubprocess;
 use crate::git::sync::process_newly_available_git_data;
-use crate::git::{delete_ref, list_refs};
+use crate::git::{delete_ref, list_refs, GitResponseBody};
 use crate::grasp06::endpoint::PrsUrl;
 use crate::grasp06::paths::prs_repo_path;
+use crate::metrics::Metrics;
 use crate::nostr::builder::Nip34WritePolicy;
 use crate::nostr::SharedDatabase;
 use crate::purgatory::promotion_hooks::NostrPurgatoryPromotionHooks;
@@ -85,9 +100,9 @@ use crate::sync::rejected_index::RejectedEventsIndex;
 ///
 /// * by the receive handler to perform `git init --bare` and register
 ///   the request as in-flight (`fetch_add` on `in_flight`),
-/// * by the receive handler again at end-of-push to decrement
-///   `in_flight` and, if no other push is in flight and the repo has
-///   zero refs, remove the bare directory,
+/// * by the detached receive-pack streaming task at end-of-push to decrement
+///   `in_flight` and, if no other push is in flight and the repo has zero refs,
+///   remove the bare directory,
 /// * by off-push cleanup paths (PR-event validation discard, purgatory
 ///   expiry) for the duration of one `delete_ref` + optional
 ///   `remove_dir_all`.
@@ -161,7 +176,8 @@ pub async fn handle_prs_receive_pack(
     git_protocol: Option<&str>,
     repo_init_locks: RepoInitLocks,
     domain: &str,
-) -> Result<Response<Full<Bytes>>, GitError> {
+    metrics: Option<Arc<Metrics>>,
+) -> Result<Response<GitResponseBody>, GitError> {
     // 1. Pre-scan refs and reject the whole push if any ref name is not
     //    `refs/nostr/<64-lowercase-hex>`. We use the same parser as the
     //    standard receive-pack path so behaviour stays in lock-step.
@@ -172,6 +188,7 @@ pub async fn handle_prs_receive_pack(
             prs.submitter.to_hex(),
             prs.identifier
         );
+        record_git_operation(&metrics, "push", "error");
         return Ok(build_git_protocol_error_response(
             GitService::ReceivePack,
             "no ref updates found in push",
@@ -187,6 +204,7 @@ pub async fn handle_prs_receive_pack(
                 prs.identifier,
                 reason
             );
+            record_git_operation(&metrics, "push", "error");
             return Ok(build_git_protocol_error_response(
                 GitService::ReceivePack,
                 &format!(
@@ -226,6 +244,7 @@ pub async fn handle_prs_receive_pack(
                     prs.identifier,
                     reason
                 );
+                record_git_operation(&metrics, "push", "error");
                 return Ok(build_git_protocol_error_response(
                     GitService::ReceivePack,
                     &format!("GRASP-06: {}", reason),
@@ -258,128 +277,82 @@ pub async fn handle_prs_receive_pack(
                 repo_path.display(),
                 e
             );
+            record_git_operation(&metrics, "push", "error");
             return Err(e);
         }
         state.in_flight.fetch_add(1, Ordering::Relaxed);
     }
 
-    // 4 + 5: run receive-pack and re-validate refs without the per-path
-    //        mutex held. Wrapping in an async block lets us catch every
-    //        exit path with the same end-of-push cleanup below.
-    let process_result: Result<Response<Full<Bytes>>, GitError> = async {
-        let response = run_receive_pack(&repo_path, &request_body, git_protocol).await?;
-
-        // If the push itself failed with a protocol error (e.g. a stale
-        // OID or a corrupt pack) we return that ERR pkt-line straight
-        // back to the client without doing any post-push validation. The
-        // pre-scan in step 1 already gated ref-name shape, so we only
-        // land here for git-level failures.
-        if response.status() != StatusCode::OK {
-            return Ok(response);
-        }
-
-        // 5. Race safety net. The pre-validation in step 2 was performed
-        //    *before* `git-receive-pack` ran, so an event with one of the
-        //    pushed ids may have arrived via WebSocket during the
-        //    receive-pack window. Re-run the same shared check now and
-        //    delete the ref on any mismatch. The `Unknown` branch is the
-        //    common case — neither DB nor purgatory have heard of this
-        //    event yet — and creates a scoped placeholder so the
-        //    30-minute purgatory sweep can clean it up if the event
-        //    never arrives.
-        let post_push_constraints = PrsUrlConstraints {
-            submitter: &prs.submitter,
-            identifier: &prs.identifier,
-            domain,
-        };
-        for (_, new_oid, ref_name) in &pushed_refs {
-            let event_id_hex = ref_name
-                .strip_prefix("refs/nostr/")
-                .expect("ref shape validated above");
-            post_push_validate(
-                &database,
-                &purgatory,
-                &repo_path,
-                post_push_constraints,
-                event_id_hex,
-                new_oid,
-                ref_name,
-            )
-            .await;
-        }
-
-        Ok(response)
-    }
-    .await;
-
-    // 6. End-of-push cleanup. Always runs — including on receive-pack
-    //    protocol errors and `?`-propagated transport errors — so a
-    //    failed push that has just initialised an empty repo does not
-    //    leak it. Decrements `in_flight`; if no other push is in flight
-    //    and the repo has zero refs left, removes the bare directory.
-    {
-        let _g = state.mu.lock().expect("prs path mutex poisoned");
-        state.in_flight.fetch_sub(1, Ordering::Relaxed);
-        if state.in_flight.load(Ordering::Relaxed) == 0 {
-            if let Ok(refs) = list_refs(&repo_path) {
-                if refs.is_empty() {
-                    if let Err(e) = std::fs::remove_dir_all(&repo_path) {
-                        warn!(
-                            "/prs/ receive-pack: failed to clean up empty repo {}: {}",
-                            repo_path.display(),
-                            e
-                        );
-                    } else {
-                        debug!(
-                            "/prs/ receive-pack: removed empty repo {} (no refs after push)",
-                            repo_path.display()
-                        );
-                    }
-                }
-            }
-        }
-    }
-
-    let response = process_result?;
-
-    // 7. Drive the standard purgatory-release pipeline so PR events
-    //    already waiting on these commits can be promoted out of
-    //    purgatory. Only fires on a successful push, and only if the
-    //    repo still exists (it may have been removed in step 6).
-    if response.status() == StatusCode::OK && repo_path.exists() {
-        let new_oids: HashSet<String> = pushed_refs
-            .iter()
-            .filter(|(_, new_oid, _)| new_oid != "0000000000000000000000000000000000000000")
-            .map(|(_, new_oid, _)| new_oid.clone())
-            .collect();
-
-        let promotion_hooks = NostrPurgatoryPromotionHooks::git_push(
-            &write_policy,
-            &rejected_events_index,
-            Some(&relay),
-        );
-
-        if let Err(e) = process_newly_available_git_data(
-            &repo_path,
-            &new_oids,
-            &database,
-            Some(&relay),
-            &purgatory,
-            Path::new(git_data_path),
-            Some(&promotion_hooks),
-        )
-        .await
+    // 4. The pre-validation and repo creation work above must stay buffered so
+    //    rejections can be returned as a complete Git `ERR` response before any
+    //    repository state is touched. Once we start `git-receive-pack`, switch
+    //    to the same streaming shape as the standard receive-pack endpoint:
+    //    write stdin here, then hand stdout/stderr and all follow-up state to a
+    //    detached task that feeds the response body channel.
+    let mut git =
+        match GitSubprocess::spawn(GitService::ReceivePack, &repo_path, false, git_protocol)
+            .map_err(GitError::ProcessSpawnFailed)
         {
-            warn!(
-                "/prs/ receive-pack: post-push processing failed for {}/{}: {}",
-                prs.submitter.to_hex(),
-                prs.identifier,
-                e
-            );
+            Ok(git) => git,
+            Err(e) => {
+                finish_prs_receive_pack(&state, &repo_path);
+                record_git_operation(&metrics, "push", "error");
+                return Err(e);
+            }
+        };
+
+    if let Some(mut stdin) = git.take_stdin() {
+        if let Err(e) = stdin.write_all(&request_body).await {
+            let _ = git.kill().await;
+            finish_prs_receive_pack(&state, &repo_path);
+            record_git_operation(&metrics, "push", "error");
+            return Err(GitError::IoError(e));
         }
+        drop(stdin);
     }
 
-    Ok(response)
+    let stdout = match git.take_stdout() {
+        Some(stdout) => stdout,
+        None => {
+            let _ = git.kill().await;
+            finish_prs_receive_pack(&state, &repo_path);
+            record_git_operation(&metrics, "push", "error");
+            return Err(GitError::IoError(io::Error::new(
+                io::ErrorKind::BrokenPipe,
+                "git receive-pack stdout unavailable",
+            )));
+        }
+    };
+    let stderr = git.take_stderr();
+    let (tx, rx) = mpsc::channel::<Result<Frame<Bytes>, io::Error>>(STREAM_CHANNEL_DEPTH);
+
+    // The request future returns as soon as the channel-backed response is
+    // built. Everything that previously happened after the buffered
+    // `run_receive_pack` call must therefore move into this task: forwarding
+    // stdout, classifying Git failures, decrementing `in_flight`, race-window
+    // validation, empty-repo cleanup, purgatory promotion, and metrics.
+    tokio::spawn(stream_prs_receive_pack_output(
+        git,
+        stdout,
+        stderr,
+        tx,
+        repo_path,
+        state,
+        pushed_refs,
+        database,
+        relay,
+        purgatory,
+        write_policy,
+        rejected_events_index,
+        git_data_path.to_string(),
+        request_body,
+        prs.submitter,
+        prs.identifier.clone(),
+        domain.to_string(),
+        metrics,
+    ));
+
+    Ok(streaming_response(GitService::ReceivePack, rx))
 }
 
 /// Return `Some(reason)` if `ref_name` is not exactly
@@ -448,75 +421,218 @@ fn ensure_repo_initialised(repo_path: &Path) -> Result<(), GitError> {
     Ok(())
 }
 
-/// Spawn `git-receive-pack`, stream stdin/stdout/stderr, and convert the
-/// result into an HTTP response — protocol errors become 200 + ERR
-/// pkt-line, transport errors bubble up as [`GitError`]. Mirrors
-/// [`crate::git::handlers::handle_receive_pack`]'s subprocess plumbing.
-async fn run_receive_pack(
-    repo_path: &Path,
-    request_body: &Bytes,
-    git_protocol: Option<&str>,
-) -> Result<Response<Full<Bytes>>, GitError> {
-    let mut git = GitSubprocess::spawn(GitService::ReceivePack, repo_path, false, git_protocol)
-        .map_err(GitError::ProcessSpawnFailed)?;
+/// End-of-push cleanup. Always runs — including on receive-pack protocol
+/// errors and client disconnects — so a failed push that has just initialised
+/// an empty repo does not leak it. Decrements `in_flight`; if no other push is
+/// in flight and the repo has zero refs left, removes the bare directory.
+fn finish_prs_receive_pack(state: &PrsPathState, repo_path: &Path) {
+    let _g = state.mu.lock().expect("prs path mutex poisoned");
+    state.in_flight.fetch_sub(1, Ordering::Relaxed);
+    if state.in_flight.load(Ordering::Relaxed) == 0 {
+        if let Ok(refs) = list_refs(repo_path) {
+            if refs.is_empty() {
+                if let Err(e) = std::fs::remove_dir_all(repo_path) {
+                    warn!(
+                        "/prs/ receive-pack: failed to clean up empty repo {}: {}",
+                        repo_path.display(),
+                        e
+                    );
+                } else {
+                    debug!(
+                        "/prs/ receive-pack: removed empty repo {} (no refs after push)",
+                        repo_path.display()
+                    );
+                }
+            }
+        }
+    }
+}
 
-    if let Some(mut stdin) = git.take_stdin() {
-        stdin
-            .write_all(request_body)
-            .await
-            .map_err(GitError::IoError)?;
-        drop(stdin);
+/// Own the live `/prs/` receive-pack after the request handler has returned its
+/// streaming response.
+///
+/// This task is the continuation of [`handle_prs_receive_pack`], not just a
+/// stdout pump. It must preserve the old buffered handler's guarantees while the
+/// HTTP body is already streaming:
+///
+/// * if stdout reaches EOF cleanly, inspect Git's exit status and stderr;
+/// * if Git failed before sending stdout, synthesize a final Git `ERR` pkt-line
+///   through the response channel;
+/// * always run `finish_prs_receive_pack` on every terminal path after
+///   `in_flight` was incremented;
+/// * only run race-window validation and purgatory promotion after a successful
+///   receive-pack.
+#[allow(clippy::too_many_arguments)]
+async fn stream_prs_receive_pack_output<S, E>(
+    mut git: GitSubprocess,
+    stdout: S,
+    stderr: Option<E>,
+    tx: mpsc::Sender<Result<Frame<Bytes>, io::Error>>,
+    repo_path: PathBuf,
+    state: Arc<PrsPathState>,
+    pushed_refs: Vec<(String, String, String)>,
+    database: SharedDatabase,
+    relay: LocalRelay,
+    purgatory: Arc<Purgatory>,
+    write_policy: Arc<Nip34WritePolicy>,
+    rejected_events_index: Arc<RejectedEventsIndex>,
+    git_data_path: String,
+    request_body: Bytes,
+    submitter: PublicKey,
+    identifier: String,
+    domain: String,
+    metrics: Option<Arc<Metrics>>,
+) where
+    S: tokio::io::AsyncRead + Unpin + Send + 'static,
+    E: tokio::io::AsyncRead + Unpin + Send + 'static,
+{
+    // Drain stderr concurrently with stdout. Git can write enough diagnostics to
+    // fill its stderr pipe while still producing stdout; draining both prevents
+    // child-process deadlock and preserves stderr for protocol-error reporting.
+    let stderr_task = stderr.map(|stderr| tokio::spawn(read_stderr_to_end(stderr)));
+    let pump_result = pump_stdout_to_channel(stdout, &tx).await;
+
+    // If the client goes away or stdout read fails, stop Git rather than letting
+    // it continue writing into a response nobody can receive. Cleanup below will
+    // still release `in_flight` after `wait()` observes process termination.
+    if !matches!(pump_result, PumpResult::Eof { .. }) {
+        let _ = git.kill().await;
     }
 
-    let mut output = Vec::new();
-    let mut stderr_output = Vec::new();
+    let status = match git.wait().await {
+        Ok(status) => status,
+        Err(e) => {
+            let _ = tx.send(Err(e)).await;
+            finish_prs_receive_pack(&state, &repo_path);
+            record_git_operation(&metrics, "push", "error");
+            return;
+        }
+    };
 
-    if let Some(stdout) = git.take_stdout() {
-        let mut stdout = stdout;
-        stdout
-            .read_to_end(&mut output)
-            .await
-            .map_err(GitError::IoError)?;
-    }
-    if let Some(stderr) = git.take_stderr() {
-        let mut stderr = stderr;
-        stderr
-            .read_to_end(&mut stderr_output)
-            .await
-            .map_err(GitError::IoError)?;
-    }
+    let stderr_output = match stderr_task {
+        Some(task) => task.await.unwrap_or_default(),
+        None => Vec::new(),
+    };
 
-    let status = git.wait().await.map_err(GitError::IoError)?;
+    let sent_stdout = match pump_result {
+        PumpResult::Eof { sent_stdout } => sent_stdout,
+        PumpResult::ClientDisconnected | PumpResult::ReadError => {
+            finish_prs_receive_pack(&state, &repo_path);
+            record_git_operation(&metrics, "push", "error");
+            return;
+        }
+    };
 
     if !status.success() {
+        record_git_operation(&metrics, "push", "error");
         let stderr_str = String::from_utf8_lossy(&stderr_output);
         if is_git_protocol_error(status.code(), &stderr_output) {
             warn!(
                 "/prs/ git-receive-pack protocol error (returning ERR pkt-line): {}",
                 stderr_str.trim()
             );
-            return Ok(build_git_protocol_error_response(
-                GitService::ReceivePack,
-                &stderr_str,
-                Some(request_body),
-            ));
+            if !sent_stdout {
+                let _ = tx
+                    .send(Ok(err_pktline_frame(
+                        GitService::ReceivePack,
+                        &stderr_str,
+                        Some(&request_body),
+                    )))
+                    .await;
+            }
+        } else {
+            error!(
+                "/prs/ git-receive-pack failed (transport): {}",
+                stderr_str.trim()
+            );
+            if !sent_stdout {
+                let msg = if stderr_str.trim().is_empty() {
+                    format!("git receive-pack failed with code {:?}", status.code())
+                } else {
+                    stderr_str.to_string()
+                };
+                let _ = tx
+                    .send(Ok(err_pktline_frame(
+                        GitService::ReceivePack,
+                        &msg,
+                        Some(&request_body),
+                    )))
+                    .await;
+            }
         }
-        error!(
-            "/prs/ git-receive-pack failed (transport): {}",
-            stderr_str.trim()
-        );
-        return Err(GitError::GitFailed(status.code()));
+        // Whether or not an ERR frame could be sent, the HTTP headers are
+        // already committed as a streaming 200 response. The only remaining
+        // safe action is to close the body after cleanup.
+        finish_prs_receive_pack(&state, &repo_path);
+        return;
     }
 
-    Ok(Response::builder()
-        .status(StatusCode::OK)
-        .header(
-            "content-type",
-            GitService::ReceivePack.result_content_type(),
+    debug!("/prs/ git-receive-pack stream completed successfully");
+    record_git_operation(&metrics, "push", "success");
+
+    // Race safety net. The pre-validation in `handle_prs_receive_pack` was
+    // performed before `git-receive-pack` ran, so an event with one of the
+    // pushed ids may have arrived via WebSocket during the receive-pack window.
+    // Re-run the same shared check now and delete the ref on any mismatch.
+    for (_, new_oid, ref_name) in &pushed_refs {
+        let event_id_hex = ref_name
+            .strip_prefix("refs/nostr/")
+            .expect("ref shape validated above");
+        let post_push_constraints = PrsUrlConstraints {
+            submitter: &submitter,
+            identifier: &identifier,
+            domain: &domain,
+        };
+        post_push_validate(
+            &database,
+            &purgatory,
+            &repo_path,
+            post_push_constraints,
+            event_id_hex,
+            new_oid,
+            ref_name,
         )
-        .header("cache-control", "no-cache")
-        .body(Full::new(Bytes::from(output)))
-        .unwrap())
+        .await;
+    }
+
+    finish_prs_receive_pack(&state, &repo_path);
+
+    // Drive the standard purgatory-release pipeline so PR events already
+    // waiting on these commits can be promoted out of purgatory. Only fires on
+    // a successful push, and only if the repo still exists (it may have been
+    // removed by end-of-push cleanup).
+    if repo_path.exists() {
+        let new_oids: HashSet<String> = pushed_refs
+            .iter()
+            .filter(|(_, new_oid, _)| new_oid != "0000000000000000000000000000000000000000")
+            .map(|(_, new_oid, _)| new_oid.clone())
+            .collect();
+
+        let promotion_hooks = NostrPurgatoryPromotionHooks::git_push(
+            write_policy,
+            rejected_events_index,
+            Some(relay.clone()),
+        );
+
+        if let Err(e) = process_newly_available_git_data(
+            &repo_path,
+            &new_oids,
+            &database,
+            Some(&relay),
+            &purgatory,
+            Path::new(&git_data_path),
+            Some(&promotion_hooks),
+        )
+        .await
+        {
+            warn!(
+                "/prs/ receive-pack: post-push processing failed for {}/{}: {}",
+                submitter.to_hex(),
+                identifier,
+                e
+            );
+        }
+    }
 }
 
 /// Race safety net for the `/prs/` receive-pack post-push phase.
@@ -534,10 +650,11 @@ async fn run_receive_pack(
 /// - was held in purgatory with a populated entry that also mismatches
 ///   (delete the ref AND drop the purgatory entry — its event is wrong).
 ///
-/// Anything left here is best-effort: errors deleting refs are logged
-/// and the push response is not changed. The end-of-push cleanup in
-/// [`handle_prs_receive_pack`] removes the bare repo if every ref ends
-/// up deleted.
+/// Anything left here is best-effort: errors deleting refs are logged and the
+/// push response is not changed. By the time this runs the response body has
+/// already streamed Git's success output, so the only safe correction is to fix
+/// repository state before `finish_prs_receive_pack` decides whether the bare
+/// repo is now empty.
 async fn post_push_validate(
     database: &SharedDatabase,
     purgatory: &Purgatory,

src/http/mod.rs

@@ -10,7 +10,7 @@ use std::pin::Pin;
 use std::sync::Arc;
 
 use base64::Engine;
-use http_body_util::{BodyExt, Full};
+use http_body_util::BodyExt;
 use hyper::body::{Bytes, Incoming};
 use hyper::header::{CONNECTION, SEC_WEBSOCKET_ACCEPT, UPGRADE};
 use hyper::server::conn::http1;
@@ -25,6 +25,7 @@ use tokio::net::TcpListener;
 
 use crate::config::Config;
 use crate::git;
+use crate::git::{empty_body, full_body, GitResponseBody};
 use crate::grasp06::receive::RepoInitLocks;
 use crate::metrics::Metrics;
 use crate::nostr::builder::Nip34WritePolicy;
@@ -34,6 +35,8 @@ use crate::purgatory::promotion_hooks::NostrPurgatoryPromotionHooks;
 use crate::purgatory::Purgatory;
 use crate::sync::rejected_index::RejectedEventsIndex;
 
+type HttpBody = GitResponseBody;
+
 /// CORS headers required by GRASP-01 specification (lines 40-47)
 const CORS_ALLOW_ORIGIN: &str = "*";
 const CORS_ALLOW_METHODS: &str = "GET, POST";
@@ -159,7 +162,7 @@ impl HttpService {
 }
 
 impl Service<Request<Incoming>> for HttpService {
-    type Response = Response<Full<Bytes>>;
+    type Response = Response<HttpBody>;
     type Error = String;
     type Future = Pin<Box<dyn Future<Output = Result<Self::Response, Self::Error>> + Send>>;
 
@@ -183,7 +186,7 @@ impl Service<Request<Incoming>> for HttpService {
                 Ok(
                     add_cors_headers(Response::builder().header("server", "ngit-grasp"))
                         .status(204)
-                        .body(Full::new(Bytes::new()))
+                        .body(empty_body())
                         .unwrap(),
                 )
             });
@@ -245,7 +248,7 @@ impl Service<Request<Incoming>> for HttpService {
                         raw_body
                     };
 
-                    let result: Result<Response<Full<Bytes>>, git::handlers::GitError> =
+                    let result: Result<Response<HttpBody>, git::handlers::GitError> =
                         match (method_clone.as_ref(), subpath.as_str()) {
                             // GET|HEAD /info/refs?service=git-{upload,receive}-pack
                             // HEAD must mirror GET headers with no body (RFC 9110 §9.3.2)
@@ -284,16 +287,33 @@ impl Service<Request<Incoming>> for HttpService {
 
                             // POST /git-upload-pack — clone/fetch.
                             (m, "git-upload-pack") if m == Method::POST => {
+                                let streams_real_repo = crate::grasp06::paths::prs_repo_path(
+                                    std::path::Path::new(&git_data_path),
+                                    &prs.submitter.to_hex(),
+                                    &prs.identifier,
+                                )
+                                .exists();
                                 let r = crate::grasp06::fetch::handle_prs_upload_pack(
                                     &prs,
                                     &git_data_path,
                                     body_bytes,
                                     git_protocol.as_deref(),
+                                    metrics_clone.clone(),
                                 )
                                 .await;
                                 if let Some(ref m) = metrics_clone {
                                     let status = if r.is_ok() { "success" } else { "error" };
-                                    m.record_git_operation("clone", status);
+                                    if streams_real_repo && r.is_ok() {
+                                        // Real /prs/ repos delegate to the standard streaming
+                                        // upload-pack handler. The streaming task records the final
+                                        // git-upload-pack success/failure when Git exits; this
+                                        // `Ok(Response)` only means Hyper received a body stream.
+                                    } else {
+                                        // Synthesized empty repos are still buffered, so `Ok` here
+                                        // means git-upload-pack has already completed. For streamed
+                                        // setup errors, there is no task to record the failure.
+                                        m.record_git_operation("clone", status);
+                                    }
                                 }
                                 r
                             }
@@ -302,6 +322,11 @@ impl Service<Request<Incoming>> for HttpService {
                             // refs/nostr/<event-id>, reject anything else,
                             // per GRASP-06 06.md line 15.
                             (m, "git-receive-pack") if m == Method::POST => {
+                                // Metrics are recorded inside the `/prs/`
+                                // handler and its detached streaming task. For
+                                // successful setup this `await` only means the
+                                // response stream was created, not that Git has
+                                // finished receiving the push.
                                 let r = crate::grasp06::receive::handle_prs_receive_pack(
                                     &prs,
                                     body_bytes,
@@ -314,12 +339,9 @@ impl Service<Request<Incoming>> for HttpService {
                                     git_protocol.as_deref(),
                                     repo_init_locks.clone(),
                                     &config_clone.domain,
+                                    metrics_clone.clone(),
                                 )
                                 .await;
-                                if let Some(ref m) = metrics_clone {
-                                    let status = if r.is_ok() { "success" } else { "error" };
-                                    m.record_git_operation("push", status);
-                                }
                                 r
                             }
 
@@ -332,7 +354,7 @@ impl Service<Request<Incoming>> for HttpService {
                             // RFC 9110 §9.3.2: HEAD response must have same headers as GET
                             // but no body.
                             let body = if method_clone == Method::HEAD {
-                                Full::new(Bytes::new())
+                                empty_body()
                             } else {
                                 body
                             };
@@ -360,7 +382,7 @@ impl Service<Request<Incoming>> for HttpService {
                             let error_msg = format!("Git error: {}", e);
                             Ok(add_cors_headers(Response::builder())
                                 .status(e.status_code())
-                                .body(Full::new(Bytes::from(error_msg)))
+                                .body(full_body(error_msg))
                                 .unwrap())
                         }
                     }
@@ -482,7 +504,7 @@ impl Service<Request<Incoming>> for HttpService {
 
                     // POST /git-upload-pack (clone/fetch)
                     (m, "git-upload-pack") if m == Method::POST => {
-                        let _repo_lifecycle_guard = match PublicKey::parse(&npub) {
+                        let repo_lifecycle_guard = match PublicKey::parse(&npub) {
                             Ok(owner_pk) => Some(
                                 lifecycle
                                     .read_repository(&owner_pk.to_hex(), &identifier)
@@ -494,11 +516,17 @@ impl Service<Request<Incoming>> for HttpService {
                             repo_path,
                             body_bytes,
                             git_protocol.as_deref(),
+                            repo_lifecycle_guard,
+                            metrics_clone.clone(),
                         )
                         .await;
                         if let Some(ref m) = metrics_clone {
-                            let status = if result.is_ok() { "success" } else { "error" };
-                            m.record_git_operation("clone", status);
+                            if result.is_err() {
+                                // On `Ok(Response)`, the streaming task owns the actual
+                                // git-upload-pack outcome metric. The synchronous handler result
+                                // only says whether stream setup succeeded.
+                                m.record_git_operation("clone", "error");
+                            }
                         }
                         result
                     }
@@ -516,7 +544,7 @@ impl Service<Request<Incoming>> for HttpService {
                                 }
                                 return Ok(add_cors_headers(Response::builder())
                                     .status(hyper::StatusCode::BAD_REQUEST)
-                                    .body(Full::new(Bytes::from(format!("Invalid npub: {}", e))))
+                                    .body(full_body(format!("Invalid npub: {}", e)))
                                     .unwrap());
                             }
                         };
@@ -528,9 +556,9 @@ impl Service<Request<Incoming>> for HttpService {
                         );
 
                         let promotion_hooks = NostrPurgatoryPromotionHooks::git_push(
-                            &write_policy,
-                            &rejected_events_index,
-                            Some(&relay),
+                            write_policy.clone(),
+                            rejected_events_index.clone(),
+                            Some(relay.clone()),
                         );
 
                         let result = git::handlers::handle_receive_pack(
@@ -544,13 +572,19 @@ impl Service<Request<Incoming>> for HttpService {
                             &git_data_path,
                             git_protocol.as_deref(),
                             repo_lifecycle_guard,
-                            Some(&promotion_hooks),
+                            Some(Arc::new(promotion_hooks)),
+                            metrics_clone.clone(),
                         )
                         .await;
 
                         if let Some(ref m) = metrics_clone {
-                            let status = if result.is_ok() { "success" } else { "error" };
-                            m.record_git_operation("push", status);
+                            if result.is_err() {
+                                // On `Ok(Response)`, the operation may be a buffered protocol
+                                // rejection already recorded by `handle_receive_pack`, or a stream
+                                // whose final git-receive-pack outcome will be recorded by the
+                                // streaming task. Do not treat this setup result as success.
+                                m.record_git_operation("push", "error");
+                            }
                         }
 
                         result
@@ -566,7 +600,7 @@ impl Service<Request<Incoming>> for HttpService {
                         // RFC 9110 §9.3.2: HEAD response must have same headers as GET
                         // but no body.
                         let body = if method == Method::HEAD {
-                            Full::new(Bytes::new())
+                            empty_body()
                         } else {
                             body
                         };
@@ -595,7 +629,7 @@ impl Service<Request<Incoming>> for HttpService {
                         let error_msg = format!("Git error: {}", e);
                         Ok(add_cors_headers(Response::builder())
                             .status(e.status_code())
-                            .body(Full::new(Bytes::from(error_msg)))
+                            .body(full_body(error_msg))
                             .unwrap())
                     }
                 }
@@ -625,7 +659,7 @@ impl Service<Request<Incoming>> for HttpService {
                         add_cors_headers(Response::builder().header("server", "ngit-grasp"))
                             .status(200)
                             .header("content-type", "application/nostr+json")
-                            .body(Full::new(Bytes::from(json)))
+                            .body(full_body(json))
                             .unwrap(),
                     )
                 });
@@ -656,7 +690,7 @@ impl Service<Request<Incoming>> for HttpService {
                         add_cors_headers(Response::builder().header("server", "ngit-grasp"))
                             .status(200)
                             .header("content-type", "text/html; charset=utf-8")
-                            .body(Full::new(Bytes::from(html)))
+                            .body(full_body(html))
                             .unwrap(),
                     )
                 } else {
@@ -666,7 +700,7 @@ impl Service<Request<Incoming>> for HttpService {
                         add_cors_headers(Response::builder().header("server", "ngit-grasp"))
                             .status(404)
                             .header("content-type", "text/html; charset=utf-8")
-                            .body(Full::new(Bytes::from(html)))
+                            .body(full_body(html))
                             .unwrap(),
                     )
                 }
@@ -722,7 +756,7 @@ impl Service<Request<Incoming>> for HttpService {
                         .header(CONNECTION, "upgrade")
                         .header(UPGRADE, "websocket")
                         .header(SEC_WEBSOCKET_ACCEPT, derived.unwrap())
-                        .body(Full::new(Bytes::new()))
+                        .body(empty_body())
                         .unwrap())
                 });
             }
@@ -738,7 +772,7 @@ impl Service<Request<Incoming>> for HttpService {
                         add_cors_headers(Response::builder().header("server", "ngit-grasp"))
                             .status(200)
                             .header("content-type", "text/plain; version=0.0.4; charset=utf-8")
-                            .body(Full::new(Bytes::from(output)))
+                            .body(full_body(output))
                             .unwrap(),
                     )
                 });
@@ -748,7 +782,7 @@ impl Service<Request<Incoming>> for HttpService {
                     Ok(
                         add_cors_headers(Response::builder().header("server", "ngit-grasp"))
                             .status(404)
-                            .body(Full::new(Bytes::from("Metrics disabled")))
+                            .body(full_body("Metrics disabled"))
                             .unwrap(),
                     )
                 });
@@ -763,7 +797,7 @@ impl Service<Request<Incoming>> for HttpService {
                         .status(200)
                         .header("content-type", "image/png")
                         .header("cache-control", "public, max-age=86400")
-                        .body(Full::new(Bytes::from_static(ICON_PNG)))
+                        .body(full_body(Bytes::from_static(ICON_PNG)))
                         .unwrap(),
                 )
             });
@@ -779,7 +813,7 @@ impl Service<Request<Incoming>> for HttpService {
                     add_cors_headers(Response::builder().header("server", "ngit-grasp"))
                         .status(200)
                         .header("content-type", "text/html; charset=utf-8")
-                        .body(Full::new(Bytes::from(html)))
+                        .body(full_body(html))
                         .unwrap(),
                 )
             } else {
@@ -789,7 +823,7 @@ impl Service<Request<Incoming>> for HttpService {
                     add_cors_headers(Response::builder().header("server", "ngit-grasp"))
                         .status(404)
                         .header("content-type", "text/html; charset=utf-8")
-                        .body(Full::new(Bytes::from(html)))
+                        .body(full_body(html))
                         .unwrap(),
                 )
             }

src/metrics/mod.rs

@@ -398,7 +398,12 @@ impl Metrics {
 
     // === Git Operation Recording Methods ===
 
-    /// Record a git operation completion
+    /// Record a git operation event.
+    ///
+    /// Buffered handlers record completion from the synchronous request path.
+    /// Streaming handlers record the final `success`/`error` from the background
+    /// task that owns the Git subprocess; their synchronous `Ok(Response)` only
+    /// means stream setup succeeded.
     pub fn record_git_operation(&self, operation: &str, status: &str) {
         self.inner
             .git_operations_total

src/purgatory/promotion_hooks.rs

@@ -21,19 +21,19 @@ use crate::nostr::lifecycle::DeletionService;
 use crate::nostr::persistence::SaveContext;
 use crate::sync::rejected_index::{EventType, RejectedEventsIndex};
 
-pub struct NostrPurgatoryPromotionHooks<'a> {
-    deletion: &'a DeletionService,
-    write_policy: Option<&'a Nip34WritePolicy>,
-    rejected_events_index: Option<&'a Arc<RejectedEventsIndex>>,
-    local_relay: Option<&'a LocalRelay>,
+pub struct NostrPurgatoryPromotionHooks {
+    deletion: DeletionService,
+    write_policy: Option<Arc<Nip34WritePolicy>>,
+    rejected_events_index: Option<Arc<RejectedEventsIndex>>,
+    local_relay: Option<LocalRelay>,
 }
 
-impl<'a> NostrPurgatoryPromotionHooks<'a> {
+impl NostrPurgatoryPromotionHooks {
     /// Run deletion recovery and accepted-event persistence hooks only.
-    pub fn recovery_only(write_policy: &'a Nip34WritePolicy) -> Self {
+    pub fn recovery_only(write_policy: &Nip34WritePolicy) -> Self {
         Self {
-            deletion: write_policy.deletion(),
-            write_policy: Some(write_policy),
+            deletion: write_policy.deletion().clone(),
+            write_policy: Some(Arc::new(write_policy.clone())),
             rejected_events_index: None,
             local_relay: None,
         }
@@ -41,12 +41,12 @@ impl<'a> NostrPurgatoryPromotionHooks<'a> {
 
     /// Git-push path: deletion recovery plus immediate hot-cache reprocessing.
     pub fn git_push(
-        write_policy: &'a Nip34WritePolicy,
-        rejected_events_index: &'a Arc<RejectedEventsIndex>,
-        local_relay: Option<&'a LocalRelay>,
+        write_policy: Arc<Nip34WritePolicy>,
+        rejected_events_index: Arc<RejectedEventsIndex>,
+        local_relay: Option<LocalRelay>,
     ) -> Self {
         Self {
-            deletion: write_policy.deletion(),
+            deletion: write_policy.deletion().clone(),
             write_policy: Some(write_policy),
             rejected_events_index: Some(rejected_events_index),
             local_relay,
@@ -55,7 +55,7 @@ impl<'a> NostrPurgatoryPromotionHooks<'a> {
 }
 
 #[async_trait]
-impl PurgatoryPromotionHooks for NostrPurgatoryPromotionHooks<'_> {
+impl PurgatoryPromotionHooks for NostrPurgatoryPromotionHooks {
     async fn before_event_saved(&self, event: &Event, context: PurgatorySaveContext) {
         tracing::trace!(
             event_id = %event.id,
@@ -77,9 +77,9 @@ impl PurgatoryPromotionHooks for NostrPurgatoryPromotionHooks<'_> {
 
     async fn after_announcement_saved(&self, event: &Event) {
         let (Some(write_policy), Some(rejected_events_index), Some(relay)) = (
-            self.write_policy,
-            self.rejected_events_index,
-            self.local_relay,
+            self.write_policy.as_deref(),
+            self.rejected_events_index.as_ref(),
+            self.local_relay.as_ref(),
         ) else {
             return;
         };

tests/git_response_streaming.rs

@@ -0,0 +1,300 @@
+//! Integration coverage for Git Smart HTTP response streaming.
+
+use std::path::Path;
+use std::sync::Arc;
+use std::time::Duration;
+
+use clap::Parser;
+use http_body_util::BodyExt;
+use hyper::body::{Bytes, Frame};
+use ngit_grasp::config::Config;
+use ngit_grasp::git::handlers::handle_receive_pack;
+use ngit_grasp::grasp06::endpoint::PrsUrl;
+use ngit_grasp::grasp06::paths::prs_repo_path;
+use ngit_grasp::grasp06::receive::{handle_prs_receive_pack, new_repo_init_locks};
+use ngit_grasp::nostr::builder::Nip34WritePolicy;
+use ngit_grasp::nostr::lifecycle::{
+    HoldingStore, ReplaceableHistoryStore, RepositoryLifecycle, Tombstones,
+};
+use ngit_grasp::nostr::SharedDatabase;
+use ngit_grasp::purgatory::Purgatory;
+use ngit_grasp::sync::rejected_index::RejectedEventsIndex;
+use nostr_relay_builder::prelude::LocalRelayBuilder;
+use nostr_sdk::prelude::*;
+use tokio::time::timeout;
+
+static PATH_ENV_LOCK: tokio::sync::Mutex<()> = tokio::sync::Mutex::const_new(());
+
+struct PathOverride {
+    original: Option<std::ffi::OsString>,
+}
+
+impl PathOverride {
+    fn prepend(dir: &Path) -> Self {
+        let original = std::env::var_os("PATH");
+        let mut paths = vec![dir.to_path_buf()];
+        if let Some(existing) = original.as_ref() {
+            paths.extend(std::env::split_paths(existing));
+        }
+        let joined = std::env::join_paths(paths).expect("join PATH entries");
+        std::env::set_var("PATH", joined);
+        Self { original }
+    }
+}
+
+impl Drop for PathOverride {
+    fn drop(&mut self) {
+        if let Some(original) = self.original.take() {
+            std::env::set_var("PATH", original);
+        } else {
+            std::env::remove_var("PATH");
+        }
+    }
+}
+
+#[tokio::test]
+async fn receive_pack_response_streams_stdout_before_subprocess_exit() {
+    let _env_lock = PATH_ENV_LOCK.lock().await;
+
+    let fake_bin = tempfile::tempdir().expect("fake git bin tempdir");
+    write_fake_git(fake_bin.path());
+    let _path = PathOverride::prepend(fake_bin.path());
+
+    let repo = tempfile::tempdir().expect("repo tempdir");
+    let git_data = tempfile::tempdir().expect("git data tempdir");
+    let database: SharedDatabase = Arc::new(nostr_memory::MemoryDatabase::unbounded());
+    let relay = LocalRelayBuilder::default().build();
+    let purgatory = Arc::new(Purgatory::new(git_data.path().to_path_buf()));
+    let owner_pubkey = "0".repeat(64);
+    let request_body = receive_pack_request_body();
+
+    let response = handle_receive_pack(
+        repo.path().to_path_buf(),
+        request_body,
+        database,
+        relay,
+        "streaming-test",
+        &owner_pubkey,
+        purgatory,
+        git_data.path().to_str().expect("utf-8 temp path"),
+        None,
+        None,
+        None,
+        None,
+    )
+    .await
+    .expect("receive-pack handler should start fake subprocess");
+
+    let mut body = response.into_body();
+
+    let first = timeout(Duration::from_secs(1), body.frame())
+        .await
+        .expect("first stdout chunk should arrive before fake git exits")
+        .expect("body should still be open")
+        .expect("first frame should not be an HTTP body error");
+    assert_eq!(frame_data(first), Bytes::from_static(b"first-progress\n"));
+
+    let no_second_yet = timeout(Duration::from_millis(250), body.frame()).await;
+    assert!(
+        no_second_yet.is_err(),
+        "body produced another frame while fake git was still sleeping; \
+         this test needs the first frame to be observed before subprocess EOF"
+    );
+
+    let second = timeout(Duration::from_secs(3), body.frame())
+        .await
+        .expect("second stdout chunk should arrive after fake git wakes")
+        .expect("body should still be open for second chunk")
+        .expect("second frame should not be an HTTP body error");
+    assert_eq!(frame_data(second), Bytes::from_static(b"second-progress\n"));
+}
+
+#[tokio::test]
+async fn prs_receive_pack_streams_stdout_before_cleanup_removes_empty_repo() {
+    let _env_lock = PATH_ENV_LOCK.lock().await;
+
+    let fake_bin = tempfile::tempdir().expect("fake git bin tempdir");
+    write_fake_git(fake_bin.path());
+    let _path = PathOverride::prepend(fake_bin.path());
+
+    let keys = Keys::generate();
+    let git_data = tempfile::tempdir().expect("git data tempdir");
+    let database: SharedDatabase = Arc::new(nostr_memory::MemoryDatabase::unbounded());
+    let relay = LocalRelayBuilder::default().build();
+    let purgatory = Arc::new(Purgatory::new(git_data.path().to_path_buf()));
+    let repo_init_locks = new_repo_init_locks();
+    let write_policy = Arc::new(test_write_policy(
+        database.clone(),
+        purgatory.clone(),
+        repo_init_locks.clone(),
+        git_data.path(),
+    ));
+    let rejected_events_index = Arc::new(RejectedEventsIndex::new(
+        Duration::from_secs(120),
+        Duration::from_secs(7 * 24 * 60 * 60),
+    ));
+    let prs = PrsUrl {
+        submitter: keys.public_key(),
+        identifier: "prs-streaming-cleanup".to_string(),
+        subpath: "git-receive-pack".to_string(),
+    };
+    let repo_path = prs_repo_path(git_data.path(), &prs.submitter.to_hex(), &prs.identifier);
+
+    let response = handle_prs_receive_pack(
+        &prs,
+        receive_pack_request_body(),
+        database,
+        relay,
+        purgatory,
+        write_policy,
+        rejected_events_index,
+        git_data.path().to_str().expect("utf-8 temp path"),
+        None,
+        repo_init_locks,
+        "streaming-test.example",
+        None,
+    )
+    .await
+    .expect("/prs/ receive-pack handler should start fake subprocess");
+
+    assert!(
+        repo_path.exists(),
+        "/prs/ repo should exist while fake receive-pack is in flight"
+    );
+
+    let mut body = response.into_body();
+
+    let first = timeout(Duration::from_secs(1), body.frame())
+        .await
+        .expect("first /prs/ stdout chunk should arrive before fake git exits")
+        .expect("/prs/ body should still be open")
+        .expect("first /prs/ frame should not be an HTTP body error");
+    assert_eq!(frame_data(first), Bytes::from_static(b"first-progress\n"));
+
+    assert!(
+        repo_path.exists(),
+        "/prs/ cleanup must not remove the repo before receive-pack exits"
+    );
+
+    let no_second_yet = timeout(Duration::from_millis(250), body.frame()).await;
+    assert!(
+        no_second_yet.is_err(),
+        "/prs/ body produced another frame while fake git was still sleeping; \
+         this test needs the first frame to be observed before subprocess EOF"
+    );
+
+    let second = timeout(Duration::from_secs(3), body.frame())
+        .await
+        .expect("second /prs/ stdout chunk should arrive after fake git wakes")
+        .expect("/prs/ body should still be open for second chunk")
+        .expect("second /prs/ frame should not be an HTTP body error");
+    assert_eq!(frame_data(second), Bytes::from_static(b"second-progress\n"));
+
+    let eof = timeout(Duration::from_secs(1), body.frame())
+        .await
+        .expect("/prs/ body should close after cleanup");
+    assert!(
+        eof.is_none(),
+        "/prs/ body should be closed after fake git exits"
+    );
+
+    assert!(
+        !repo_path.exists(),
+        "/prs/ cleanup should remove zero-ref repo after receive-pack exits"
+    );
+}
+
+fn frame_data(frame: Frame<Bytes>) -> Bytes {
+    frame.into_data().expect("frame should contain data")
+}
+
+fn receive_pack_request_body() -> Bytes {
+    let old_oid = "0".repeat(40);
+    let new_oid = "1".repeat(40);
+    let event_id = "a".repeat(64);
+    let mut payload = Vec::new();
+    payload.extend_from_slice(format!("{old_oid} {new_oid} refs/nostr/{event_id}").as_bytes());
+    payload.push(0);
+    payload.extend_from_slice(b"report-status side-band-64k\n");
+
+    let mut request = Vec::new();
+    request.extend_from_slice(format!("{:04x}", payload.len() + 4).as_bytes());
+    request.extend_from_slice(&payload);
+    request.extend_from_slice(b"0000");
+    Bytes::from(request)
+}
+
+fn test_write_policy(
+    database: SharedDatabase,
+    purgatory: Arc<Purgatory>,
+    repo_init_locks: ngit_grasp::grasp06::receive::RepoInitLocks,
+    git_data_path: &Path,
+) -> Nip34WritePolicy {
+    let config = Config::parse_from([
+        "ngit-grasp-test",
+        "--domain",
+        "streaming-test.example",
+        "--grasp06-enable",
+    ]);
+
+    Nip34WritePolicy::new(
+        database,
+        Tombstones::in_memory(),
+        HoldingStore::in_memory(),
+        RepositoryLifecycle::in_memory(),
+        ReplaceableHistoryStore::in_memory(),
+        git_data_path.to_path_buf(),
+        purgatory,
+        config,
+        repo_init_locks,
+    )
+}
+
+fn write_fake_git(bin_dir: &Path) {
+    let git_path = bin_dir.join("git");
+    std::fs::write(
+        &git_path,
+        r#"#!/usr/bin/env bash
+set -euo pipefail
+
+is_receive_pack=0
+for arg in "$@"; do
+  if [ "$arg" = "receive-pack" ]; then
+    is_receive_pack=1
+  fi
+done
+
+if [ "$is_receive_pack" = "1" ]; then
+cat >/dev/null
+printf 'first-progress\n'
+sleep 2
+printf 'second-progress\n'
+exit 0
+fi
+
+if [ "$1" = "init" ]; then
+  repo="${@: -1}"
+  mkdir -p "$repo"
+  exit 0
+fi
+
+if [ "$1" = "for-each-ref" ]; then
+  exit 0
+fi
+
+echo "fake git only supports init, for-each-ref, and receive-pack" >&2
+exit 1
+"#,
+    )
+    .expect("write fake git executable");
+
+    #[cfg(unix)]
+    {
+        use std::os::unix::fs::PermissionsExt;
+        let mut perms = std::fs::metadata(&git_path)
+            .expect("fake git metadata")
+            .permissions();
+        perms.set_mode(0o755);
+        std::fs::set_permissions(&git_path, perms).expect("chmod fake git");
+    }
+}