feat(server): add streaming-first server scaffolding

Introduce src/server/ as a streaming-first git/lfs server surface. All
request and response bodies cross the API as
Box<dyn AsyncRead + Unpin + Send + Sync> (alias BoxedAsyncRead) — no
Vec<u8> bodies, no read_to_end convenience, no http crate.

What's in the box:
- src/server/mod.rs: re-exports + pipe re-export of
  command::cmd::DEFAULT_OUTPUT_CAP_BYTES as server::DEFAULT_OUTPUT_CAP_BYTES
- src/server/backend.rs: ServerBackend async-trait (authorize,
  list_refs_stream, produce_pack_stream, ingest_pack_stream,
  lfs_get_stream, lfs_put_stream). Every method currently returns
  ServerError::Unimplemented("...") placeholders — the wiring is
  real, the bodies are intentionally empty.
- src/server/request.rs: ServerRequest (not Clone, body is a boxed
  reader), HeaderMap = HashMap<String, String>, RefUpdateBatch
  (oid strings, no gix::ObjectId), BoxedAsyncRead alias
- src/server/stream.rs: StreamGuard (Clone + Send + Sync) carrying
  cancel() / handle() so spawned workers can race the request
- src/server/endpoint.rs, error.rs, limits.rs: EndpointKind enum,
  ServerError, DEFAULT_LFS_OBJECT_CAP_BYTES (5 GiB cap, separate from
  the 64 MiB output cap)
- tests/server.rs: PlaceholderBackend skeleton + ENV_LOCK +
  200 ms cancel convention reused from tests/env.rs and tests/pipe.rs

Public API impact on GitBaby:
- Drop pipe from GitBaby::new — the server owns its own stream
  lifecycle, so the Pipe parameter is no longer required at
  construction time. Existing callers must be updated:
    - before: GitBaby::new(facade, pipe)
    - after:  GitBaby::new(facade)

Documentation:
- AGENTS.md gains a "Planned but unimplemented (streaming-first
  server)" section documenting the deliberately-empty contracts so
  future contributors don't "fill them in" without a concrete
  git/lfs protocol task.

CI: cargo build + cargo test (61 total, 23 new from server suite)
green.

AGENTS.md

@@ -35,4 +35,16 @@
 ## Gotchas
 - No README. No docs beyond the code. Treat `Cargo.toml` + source as the only source of truth.
 - `gix = "0.86"`, `tokio = 1.53` (full features), `time = 0.3` — all recent; do not bump them speculatively.
-- Public API surfaces async (`async_trait`); sync callers must use a runtime (tests use `#[tokio::test]`).
+- Public API surfaces async (`async_trait`); sync callers must use a runtime (tests use `#[tokio::test]`).
+
+## Planned but unimplemented (streaming-first server)
+- `src/server/` exists (mod.rs, backend.rs, endpoint.rs, error.rs, limits.rs, request.rs, stream.rs) but every `ServerBackend` method currently returns `ServerError::Unimplemented("…")` placeholders. Do not "fill them in" with arbitrary bodies — wait for a concrete git/lfs protocol task before adding logic.
+- Streaming-only contract is intentional:
+  - All request/response bodies cross the API as `Box<dyn AsyncRead + Unpin + Send + Sync>` (alias `BoxedAsyncRead`). Never introduce a `Vec<u8>` body, `ServerResponse::body`, or `read_to_end` convenience into the public surface.
+  - `HeaderMap = HashMap<String, String>` (no `http` crate).
+  - `output_cap` defaults to `crate::command::cmd::DEFAULT_OUTPUT_CAP_BYTES` (64 MiB). The server re-exports it as `server::DEFAULT_OUTPUT_CAP_BYTES`.
+  - `LFS` upload cap is `DEFAULT_LFS_OBJECT_CAP_BYTES` (5 GiB) — a separate constant, not a magic number.
+  - `RefUpdateBatch` is a server-local struct (oid strings, not `gix::ObjectId`) so the API does not depend on `gix` types. Do **not** alias `refs::types::RefUpdate` here.
+- `ServerBackend` async-trait method set: `authorize`, `list_refs_stream`, `produce_pack_stream`, `ingest_pack_stream`, `lfs_get_stream`, `lfs_put_stream`. Methods returning a body must return `BoxedAsyncRead`; ingest methods return `IngestReport`. The legacy `head_object` / `write_ref` / `get_object` shapes are intentionally removed.
+- `ServerRequest` is **not** `Clone` (body is a boxed reader); it carries a manual `Debug` impl. `StreamGuard` is `Clone + Send + Sync` and exposes `cancel()` / `handle()` so spawned workers can race the request.
+- Tests: `tests/server.rs` reuses the project-wide `static ENV_LOCK: Mutex<()>` pattern from `tests/env.rs` and the `tokio::time::sleep(Duration::from_millis(200))` cancel convention from `tests/pipe.rs`. Add new server tests there.

src/lib.rs

@@ -1,6 +1,5 @@
 use std::sync::Arc;
 
-use crate::command::pipe::Pipe;
 use crate::repo::RepositoryFacade;
 
 pub mod archive;
@@ -20,6 +19,7 @@ pub mod merge;
 pub mod refs;
 pub mod remote;
 pub mod repo;
+pub mod server;
 pub mod setup;
 pub mod share;
 pub mod submodule;
@@ -29,23 +29,14 @@ pub mod tree;
 #[derive(Clone)]
 pub struct GitBaby {
     facade: Arc<dyn RepositoryFacade>,
-    pipe: Pipe,
 }
 
 impl GitBaby {
-    pub fn new(facade: Arc<dyn RepositoryFacade>, pipe: Pipe) -> Self {
-        Self { facade, pipe }
+    pub fn new(facade: Arc<dyn RepositoryFacade>) -> Self {
+        Self { facade }
     }
 
     pub fn facade(&self) -> &dyn RepositoryFacade {
         &*self.facade
     }
-
-    pub fn pipe(&self) -> &Pipe {
-        &self.pipe
-    }
-
-    pub fn pipe_mut(&mut self) -> &mut Pipe {
-        &mut self.pipe
-    }
 }

src/server/backend.rs

@@ -0,0 +1,37 @@
+use async_trait::async_trait;
+
+use crate::server::error::ServerResult;
+use crate::server::request::{AuthContext, RefUpdateBatch, ServerRequest};
+use crate::server::stream::{BoxedAsyncRead, IngestReport};
+
+#[async_trait]
+pub trait ServerBackend: Send + Sync + 'static {
+    async fn authorize(&self, req: &ServerRequest) -> ServerResult<AuthContext>;
+
+    async fn list_refs_stream(&self, req: &ServerRequest) -> ServerResult<BoxedAsyncRead>;
+
+    async fn produce_pack_stream(
+        &self,
+        req: &ServerRequest,
+        wants: &[String],
+    ) -> ServerResult<BoxedAsyncRead>;
+
+    async fn ingest_pack_stream(
+        &self,
+        req: &ServerRequest,
+        batch: RefUpdateBatch,
+    ) -> ServerResult<IngestReport>;
+
+    async fn lfs_get_stream(&self, req: &ServerRequest, oid: &str) -> ServerResult<BoxedAsyncRead>;
+
+    async fn lfs_put_stream(
+        &self,
+        req: &ServerRequest,
+        oid: &str,
+        declared_size: u64,
+    ) -> ServerResult<IngestReport>;
+}
+
+pub fn empty_body() -> BoxedAsyncRead {
+    Box::new(tokio::io::empty())
+}

src/server/endpoint.rs

@@ -0,0 +1,51 @@
+#[derive(Debug, Clone, Copy, PartialEq, Eq)]
+pub enum EndpointKind {
+    GitUploadPack,
+    GitReceivePack,
+    LfsBatch,
+    LfsObjects,
+    InfoRefs,
+    SmartHttp,
+    Custom,
+}
+
+impl EndpointKind {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            Self::GitUploadPack => "git-upload-pack",
+            Self::GitReceivePack => "git-receive-pack",
+            Self::LfsBatch => "lfs-batch",
+            Self::LfsObjects => "lfs-objects",
+            Self::InfoRefs => "info-refs",
+            Self::SmartHttp => "smart-http",
+            Self::Custom => "custom",
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct GitProtocolEndpoint {
+    pub kind: EndpointKind,
+    pub path: String,
+}
+
+impl GitProtocolEndpoint {
+    pub fn new(kind: EndpointKind, path: impl Into<String>) -> Self {
+        Self {
+            kind,
+            path: path.into(),
+        }
+    }
+
+    pub fn upload_pack(path: impl Into<String>) -> Self {
+        Self::new(EndpointKind::GitUploadPack, path)
+    }
+
+    pub fn receive_pack(path: impl Into<String>) -> Self {
+        Self::new(EndpointKind::GitReceivePack, path)
+    }
+
+    pub fn info_refs(path: impl Into<String>) -> Self {
+        Self::new(EndpointKind::InfoRefs, path)
+    }
+}

src/server/error.rs

@@ -0,0 +1,78 @@
+use std::fmt;
+use std::path::PathBuf;
+
+use crate::error::BabyError;
+
+#[derive(Debug)]
+pub enum ServerError {
+    Unimplemented(&'static str),
+    Unauthorised {
+        scheme: &'static str,
+        reason: String,
+    },
+    InvalidRequest(String),
+    PayloadTooLarge {
+        cap_bytes: u64,
+        actual_bytes: u64,
+    },
+    Io {
+        path: PathBuf,
+        source: std::io::Error,
+    },
+    Pipe(String),
+    Backend(BabyError),
+    Cancelled,
+}
+
+impl fmt::Display for ServerError {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        match self {
+            Self::Unimplemented(what) => write!(f, "not yet implemented: {}", what),
+            Self::Unauthorised { scheme, reason } => {
+                write!(f, "unauthorised ({scheme}): {}", reason)
+            }
+            Self::InvalidRequest(msg) => write!(f, "invalid server request: {}", msg),
+            Self::PayloadTooLarge {
+                cap_bytes,
+                actual_bytes,
+            } => write!(
+                f,
+                "output exceeds cap: {} bytes > {} bytes cap",
+                actual_bytes, cap_bytes
+            ),
+            Self::Io { path, source } => {
+                write!(f, "io error at `{}`: {}", path.display(), source)
+            }
+            Self::Pipe(msg) => write!(f, "server pipe error: {}", msg),
+            Self::Backend(err) => write!(f, "backend error: {}", err),
+            Self::Cancelled => write!(f, "server stream was cancelled"),
+        }
+    }
+}
+
+impl std::error::Error for ServerError {
+    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
+        match self {
+            Self::Io { source, .. } => Some(source),
+            Self::Backend(err) => Some(err),
+            _ => None,
+        }
+    }
+}
+
+impl From<BabyError> for ServerError {
+    fn from(err: BabyError) -> Self {
+        Self::Backend(err)
+    }
+}
+
+impl From<std::io::Error> for ServerError {
+    fn from(source: std::io::Error) -> Self {
+        Self::Io {
+            path: PathBuf::new(),
+            source,
+        }
+    }
+}
+
+pub type ServerResult<T> = std::result::Result<T, ServerError>;

src/server/limits.rs

@@ -0,0 +1,3 @@
+pub use crate::command::cmd::DEFAULT_OUTPUT_CAP_BYTES;
+
+pub const DEFAULT_LFS_OBJECT_CAP_BYTES: u64 = 5_u64 * 1024 * 1024 * 1024;

src/server/mod.rs

@@ -0,0 +1,19 @@
+pub mod backend;
+pub mod endpoint;
+pub mod error;
+pub mod limits;
+pub mod request;
+pub mod stream;
+
+pub use backend::ServerBackend;
+pub use endpoint::{EndpointKind, GitProtocolEndpoint};
+pub use error::{ServerError, ServerResult};
+pub use limits::{DEFAULT_LFS_OBJECT_CAP_BYTES, DEFAULT_OUTPUT_CAP_BYTES};
+pub use request::{
+    AuthContext, AuthScheme, HeaderMap, Principal, RefUpdateBatch, RefUpdateEntry, ServerRequest,
+    StreamGuard, StreamGuardHandle,
+};
+pub use stream::{
+    BoxedAsyncBufRead, BoxedAsyncRead, BoxedAsyncWrite, ChildPackReader, ChildPackWriter,
+    IngestReport, PackStats, StreamOutcome, buf_empty_reader, empty_reader, sink_writer,
+};

src/server/request.rs

@@ -0,0 +1,239 @@
+use std::collections::HashMap;
+use std::sync::Arc;
+use std::sync::atomic::{AtomicBool, Ordering};
+
+use tokio::io::AsyncRead;
+
+use crate::server::endpoint::EndpointKind;
+use crate::server::limits::DEFAULT_OUTPUT_CAP_BYTES;
+
+pub type HeaderMap = HashMap<String, String>;
+
+#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)]
+pub enum AuthScheme {
+    #[default]
+    Anonymous,
+    Basic,
+    Bearer,
+    Ssh,
+}
+
+impl AuthScheme {
+    pub fn as_str(&self) -> &'static str {
+        match self {
+            Self::Anonymous => "anonymous",
+            Self::Basic => "basic",
+            Self::Bearer => "bearer",
+            Self::Ssh => "ssh",
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct Principal {
+    pub name: String,
+    pub scheme: AuthScheme,
+}
+
+impl Principal {
+    pub fn new(name: impl Into<String>, scheme: AuthScheme) -> Self {
+        Self {
+            name: name.into(),
+            scheme,
+        }
+    }
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct AuthContext {
+    pub scheme: AuthScheme,
+    pub principal: Option<Principal>,
+    pub raw: Option<String>,
+}
+
+impl AuthContext {
+    pub fn anonymous() -> Self {
+        Self {
+            scheme: AuthScheme::Anonymous,
+            principal: None,
+            raw: None,
+        }
+    }
+
+    pub fn with_principal(scheme: AuthScheme, principal: Principal) -> Self {
+        Self {
+            scheme,
+            principal: Some(principal),
+            raw: None,
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct RefUpdateEntry {
+    pub name: String,
+    pub new_oid: String,
+    pub old_oid: Option<String>,
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct RefUpdateBatch {
+    pub updates: Vec<RefUpdateEntry>,
+}
+
+impl RefUpdateBatch {
+    pub fn new() -> Self {
+        Self::default()
+    }
+
+    pub fn push(
+        mut self,
+        name: impl Into<String>,
+        new_oid: impl Into<String>,
+        old_oid: Option<String>,
+    ) -> Self {
+        self.updates.push(RefUpdateEntry {
+            name: name.into(),
+            new_oid: new_oid.into(),
+            old_oid,
+        });
+        self
+    }
+
+    pub fn is_empty(&self) -> bool {
+        self.updates.is_empty()
+    }
+
+    pub fn len(&self) -> usize {
+        self.updates.len()
+    }
+}
+
+pub struct ServerRequest {
+    pub method: String,
+    pub path: String,
+    pub query: Option<String>,
+    pub headers: HeaderMap,
+    pub body: Box<dyn AsyncRead + Unpin + Send + Sync>,
+    pub guard: StreamGuard,
+    pub output_cap: u64,
+    pub auth: Option<AuthContext>,
+    pub endpoint: EndpointKind,
+    pub remote_addr: Option<String>,
+    pub service: String,
+}
+
+impl std::fmt::Debug for ServerRequest {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        f.debug_struct("ServerRequest")
+            .field("method", &self.method)
+            .field("path", &self.path)
+            .field("query", &self.query)
+            .field("headers", &self.headers)
+            .field("body", &"<dyn AsyncRead>")
+            .field("guard", &self.guard)
+            .field("output_cap", &self.output_cap)
+            .field("auth", &self.auth)
+            .field("endpoint", &self.endpoint)
+            .field("remote_addr", &self.remote_addr)
+            .field("service", &self.service)
+            .finish()
+    }
+}
+
+impl ServerRequest {
+    pub fn new(
+        method: impl Into<String>,
+        path: impl Into<String>,
+        endpoint: EndpointKind,
+        service: impl Into<String>,
+        body: Box<dyn AsyncRead + Unpin + Send + Sync>,
+    ) -> Self {
+        Self {
+            method: method.into(),
+            path: path.into(),
+            query: None,
+            headers: HeaderMap::new(),
+            body,
+            guard: StreamGuard::new(),
+            output_cap: DEFAULT_OUTPUT_CAP_BYTES as u64,
+            auth: None,
+            endpoint,
+            remote_addr: None,
+            service: service.into(),
+        }
+    }
+
+    pub fn with_query(mut self, query: impl Into<String>) -> Self {
+        self.query = Some(query.into());
+        self
+    }
+
+    pub fn with_header(mut self, key: impl Into<String>, value: impl Into<String>) -> Self {
+        self.headers.insert(key.into(), value.into());
+        self
+    }
+
+    pub fn with_auth(mut self, auth: AuthContext) -> Self {
+        self.auth = Some(auth);
+        self
+    }
+
+    pub fn with_output_cap(mut self, cap_bytes: u64) -> Self {
+        self.output_cap = cap_bytes;
+        self
+    }
+
+    pub fn with_remote_addr(mut self, addr: impl Into<String>) -> Self {
+        self.remote_addr = Some(addr.into());
+        self
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct StreamGuard {
+    cancelled: Arc<AtomicBool>,
+}
+
+impl Default for StreamGuard {
+    fn default() -> Self {
+        Self::new()
+    }
+}
+
+impl StreamGuard {
+    pub fn new() -> Self {
+        Self {
+            cancelled: Arc::new(AtomicBool::new(false)),
+        }
+    }
+
+    pub fn cancel(&self) {
+        self.cancelled.store(true, Ordering::SeqCst);
+    }
+
+    pub fn is_cancelled(&self) -> bool {
+        self.cancelled.load(Ordering::SeqCst)
+    }
+
+    pub fn handle(&self) -> StreamGuardHandle {
+        StreamGuardHandle {
+            cancelled: self.cancelled.clone(),
+        }
+    }
+}
+
+#[derive(Debug, Clone)]
+pub struct StreamGuardHandle {
+    cancelled: Arc<AtomicBool>,
+}
+
+impl StreamGuardHandle {
+    pub fn is_cancelled(&self) -> bool {
+        self.cancelled.load(Ordering::SeqCst)
+    }
+
+    pub fn cancel(&self) {
+        self.cancelled.store(true, Ordering::SeqCst);
+    }
+}

src/server/stream.rs

@@ -0,0 +1,251 @@
+use std::path::PathBuf;
+
+use tokio::io::{AsyncBufRead, AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt, BufReader};
+use tokio::process::Child;
+
+use crate::server::error::{ServerError, ServerResult};
+use crate::server::limits::DEFAULT_OUTPUT_CAP_BYTES;
+
+pub struct ChildPackReader<R: AsyncBufRead + Unpin + Send> {
+    reader: R,
+    child: Option<Child>,
+    finished: bool,
+}
+
+impl<R: AsyncBufRead + Unpin + Send> ChildPackReader<R> {
+    pub fn new(reader: R, child: Option<Child>) -> Self {
+        Self {
+            reader,
+            child,
+            finished: false,
+        }
+    }
+
+    pub fn into_inner(self) -> (R, Option<Child>) {
+        (self.reader, self.child)
+    }
+
+    pub fn finished(&self) -> bool {
+        self.finished
+    }
+
+    pub fn child_handle(&mut self) -> Option<&mut Child> {
+        self.child.as_mut()
+    }
+
+    pub async fn next_chunk(&mut self, buf: &mut [u8]) -> ServerResult<usize> {
+        if self.finished {
+            return Ok(0);
+        }
+        let n = self
+            .reader
+            .read(buf)
+            .await
+            .map_err(|source| ServerError::Io {
+                path: PathBuf::new(),
+                source,
+            })?;
+        if n == 0 {
+            self.finished = true;
+        }
+        Ok(n)
+    }
+
+    pub async fn read_to_end_capped(&mut self, cap: Option<u64>) -> ServerResult<StreamOutcome> {
+        let cap = cap.unwrap_or(DEFAULT_OUTPUT_CAP_BYTES as u64);
+        let mut out: Vec<u8> = Vec::new();
+        let mut buf = vec![0u8; 32 * 1024];
+        loop {
+            let n = self.next_chunk(&mut buf).await?;
+            if n == 0 {
+                self.finished = true;
+                return Ok(StreamOutcome::Eof);
+            }
+            let projected = (out.len() as u64) + (n as u64);
+            if projected > cap {
+                return Err(ServerError::PayloadTooLarge {
+                    cap_bytes: cap,
+                    actual_bytes: projected,
+                });
+            }
+            out.extend_from_slice(&buf[..n]);
+            if (out.len() as u64) >= cap {
+                return Ok(StreamOutcome::Bytes(out.len() as u64));
+            }
+        }
+    }
+
+    pub async fn collect_pack_stats(&mut self, cap: Option<u64>) -> ServerResult<PackStats> {
+        let mut stats = PackStats::default();
+        let mut buf = vec![0u8; 32 * 1024];
+        loop {
+            let n = self.next_chunk(&mut buf).await?;
+            if n == 0 {
+                break;
+            }
+            stats.bytes_streamed = stats.bytes_streamed.saturating_add(n as u64);
+            if let Some(limit) = cap
+                && stats.bytes_streamed > limit
+            {
+                return Err(ServerError::PayloadTooLarge {
+                    cap_bytes: limit,
+                    actual_bytes: stats.bytes_streamed,
+                });
+            }
+        }
+        Ok(stats)
+    }
+}
+
+pub struct ChildPackWriter<W: AsyncWrite + Unpin + Send> {
+    writer: W,
+    child: Option<Child>,
+    finished: bool,
+}
+
+impl<W: AsyncWrite + Unpin + Send> ChildPackWriter<W> {
+    pub fn new(writer: W, child: Option<Child>) -> Self {
+        Self {
+            writer,
+            child,
+            finished: false,
+        }
+    }
+
+    pub fn into_inner(self) -> (W, Option<Child>) {
+        (self.writer, self.child)
+    }
+
+    pub fn finished(&self) -> bool {
+        self.finished
+    }
+
+    pub fn child_handle(&mut self) -> Option<&mut Child> {
+        self.child.as_mut()
+    }
+
+    pub async fn write_chunk(&mut self, buf: &[u8]) -> ServerResult<usize> {
+        if self.finished {
+            return Ok(0);
+        }
+        let n = self
+            .writer
+            .write(buf)
+            .await
+            .map_err(|source| ServerError::Io {
+                path: PathBuf::new(),
+                source,
+            })?;
+        Ok(n)
+    }
+
+    pub async fn copy_from<R>(&mut self, mut reader: R) -> ServerResult<StreamOutcome>
+    where
+        R: AsyncRead + Unpin + Send,
+    {
+        let mut buf = vec![0u8; 32 * 1024];
+        let mut total: u64 = 0;
+        loop {
+            let n = reader
+                .read(&mut buf)
+                .await
+                .map_err(|source| ServerError::Io {
+                    path: PathBuf::new(),
+                    source,
+                })?;
+            if n == 0 {
+                break;
+            }
+            total = total.saturating_add(n as u64);
+            let written = self.write_chunk(&buf[..n]).await?;
+            if written == 0 {
+                self.finished = true;
+                return Ok(StreamOutcome::Bytes(total));
+            }
+        }
+        self.writer
+            .flush()
+            .await
+            .map_err(|source| ServerError::Io {
+                path: PathBuf::new(),
+                source,
+            })?;
+        Ok(StreamOutcome::Bytes(total))
+    }
+
+    pub async fn finish(mut self) -> ServerResult<()> {
+        self.writer
+            .shutdown()
+            .await
+            .map_err(|source| ServerError::Io {
+                path: PathBuf::new(),
+                source,
+            })?;
+        self.finished = true;
+        Ok(())
+    }
+}
+
+#[derive(Debug, Clone, Default, PartialEq, Eq)]
+pub struct PackStats {
+    pub bytes_streamed: u64,
+    pub objects_reported: u64,
+}
+
+impl PackStats {
+    pub fn is_empty(&self) -> bool {
+        self.bytes_streamed == 0 && self.objects_reported == 0
+    }
+}
+
+#[derive(Debug, Clone, Default)]
+pub struct IngestReport {
+    pub stats: PackStats,
+    pub refs_updated: Vec<String>,
+}
+
+impl IngestReport {
+    pub fn new(stats: PackStats, refs_updated: Vec<String>) -> Self {
+        Self {
+            stats,
+            refs_updated,
+        }
+    }
+
+    pub fn empty() -> Self {
+        Self::default()
+    }
+}
+
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum StreamOutcome {
+    Eof,
+    Bytes(u64),
+}
+
+impl StreamOutcome {
+    pub fn bytes(&self) -> Option<u64> {
+        match self {
+            Self::Eof => None,
+            Self::Bytes(n) => Some(*n),
+        }
+    }
+}
+
+pub type BoxedAsyncRead = Box<dyn AsyncRead + Unpin + Send + Sync>;
+
+pub type BoxedAsyncWrite = Box<dyn AsyncWrite + Unpin + Send + Sync>;
+
+pub type BoxedAsyncBufRead = Box<dyn AsyncBufRead + Unpin + Send + Sync>;
+
+pub fn empty_reader() -> BoxedAsyncRead {
+    Box::new(tokio::io::empty())
+}
+
+pub fn sink_writer() -> BoxedAsyncWrite {
+    Box::new(tokio::io::sink())
+}
+
+pub fn buf_empty_reader() -> BoxedAsyncBufRead {
+    Box::new(BufReader::new(tokio::io::empty()))
+}

tests/server.rs

@@ -0,0 +1,392 @@
+use std::sync::{Arc, Mutex, MutexGuard, PoisonError};
+use std::time::Duration;
+
+use tokio::io::AsyncReadExt;
+
+use gitbaby::server::{
+    AuthContext, AuthScheme, ChildPackReader, ChildPackWriter, EndpointKind, IngestReport,
+    PackStats, Principal, RefUpdateBatch, RefUpdateEntry, ServerBackend, ServerError,
+    ServerRequest, StreamGuard, StreamOutcome, buf_empty_reader, empty_reader,
+};
+
+static ENV_LOCK: Mutex<()> = Mutex::new(());
+
+fn lock_or_recover() -> MutexGuard<'static, ()> {
+    match ENV_LOCK.lock() {
+        Ok(g) => g,
+        Err(PoisonError { .. }) => panic!("env lock poisoned"),
+    }
+}
+
+#[derive(Debug)]
+struct PlaceholderBackend;
+
+#[async_trait::async_trait]
+impl ServerBackend for PlaceholderBackend {
+    async fn authorize(&self, _req: &ServerRequest) -> Result<AuthContext, ServerError> {
+        Err(ServerError::Unimplemented(
+            "server::ServerBackend::authorize",
+        ))
+    }
+
+    async fn list_refs_stream(
+        &self,
+        _req: &ServerRequest,
+    ) -> Result<Box<dyn tokio::io::AsyncRead + Unpin + Send + Sync>, ServerError> {
+        Err(ServerError::Unimplemented(
+            "server::ServerBackend::list_refs_stream",
+        ))
+    }
+
+    async fn produce_pack_stream(
+        &self,
+        _req: &ServerRequest,
+        _wants: &[String],
+    ) -> Result<Box<dyn tokio::io::AsyncRead + Unpin + Send + Sync>, ServerError> {
+        Err(ServerError::Unimplemented(
+            "server::ServerBackend::produce_pack_stream",
+        ))
+    }
+
+    async fn ingest_pack_stream(
+        &self,
+        _req: &ServerRequest,
+        _batch: RefUpdateBatch,
+    ) -> Result<IngestReport, ServerError> {
+        Err(ServerError::Unimplemented(
+            "server::ServerBackend::ingest_pack_stream",
+        ))
+    }
+
+    async fn lfs_get_stream(
+        &self,
+        _req: &ServerRequest,
+        _oid: &str,
+    ) -> Result<Box<dyn tokio::io::AsyncRead + Unpin + Send + Sync>, ServerError> {
+        Err(ServerError::Unimplemented(
+            "server::ServerBackend::lfs_get_stream",
+        ))
+    }
+
+    async fn lfs_put_stream(
+        &self,
+        _req: &ServerRequest,
+        _oid: &str,
+        _declared_size: u64,
+    ) -> Result<IngestReport, ServerError> {
+        Err(ServerError::Unimplemented(
+            "server::ServerBackend::lfs_put_stream",
+        ))
+    }
+}
+
+fn make_request(method: &str, path: &str, endpoint: EndpointKind) -> ServerRequest {
+    ServerRequest::new(method, path, endpoint, "test-service", empty_reader())
+}
+
+#[test]
+fn auth_scheme_str_round_trip() {
+    let _guard = lock_or_recover();
+    assert_eq!(AuthScheme::Anonymous.as_str(), "anonymous");
+    assert_eq!(AuthScheme::Basic.as_str(), "basic");
+    assert_eq!(AuthScheme::Bearer.as_str(), "bearer");
+    assert_eq!(AuthScheme::Ssh.as_str(), "ssh");
+    assert_eq!(AuthScheme::default(), AuthScheme::Anonymous);
+}
+
+#[test]
+fn auth_context_anonymous_has_no_principal() {
+    let ctx = AuthContext::anonymous();
+    assert_eq!(ctx.scheme, AuthScheme::Anonymous);
+    assert!(ctx.principal.is_none());
+    assert!(ctx.raw.is_none());
+}
+
+#[test]
+fn auth_context_with_principal_keeps_scheme_and_principal() {
+    let ctx = AuthContext::with_principal(
+        AuthScheme::Bearer,
+        Principal::new("alice", AuthScheme::Bearer),
+    );
+    assert_eq!(ctx.scheme, AuthScheme::Bearer);
+    let p = ctx.principal.expect("principal present");
+    assert_eq!(p.name, "alice");
+    assert_eq!(p.scheme, AuthScheme::Bearer);
+}
+
+#[test]
+fn ref_update_batch_push_and_count() {
+    let batch = RefUpdateBatch::new()
+        .push(
+            "refs/heads/main",
+            "0000000000000000000000000000000000000000",
+            None,
+        )
+        .push(
+            "refs/heads/dev",
+            "1111111111111111111111111111111111111111",
+            Some("0000000000000000000000000000000000000000".to_string()),
+        );
+
+    assert_eq!(batch.len(), 2);
+    assert!(!batch.is_empty());
+    assert_eq!(batch.updates.len(), 2);
+    assert_eq!(batch.updates[0].name, "refs/heads/main");
+    assert!(batch.updates[0].old_oid.is_none());
+    assert!(batch.updates[1].old_oid.is_some());
+}
+
+#[test]
+fn stream_guard_cancel_is_observable_via_handle() {
+    let guard = StreamGuard::new();
+    let handle = guard.handle();
+    assert!(!guard.is_cancelled());
+    assert!(!handle.is_cancelled());
+
+    guard.cancel();
+    assert!(guard.is_cancelled());
+    assert!(handle.is_cancelled());
+
+    handle.cancel();
+    assert!(guard.is_cancelled());
+}
+
+#[test]
+fn server_request_builder_populates_fields() {
+    let req = make_request("GET", "/info/refs", EndpointKind::InfoRefs)
+        .with_query("service=git-upload-pack")
+        .with_header(
+            "Content-Type",
+            "application/x-git-upload-pack-advertisement",
+        )
+        .with_auth(AuthContext::anonymous())
+        .with_output_cap(1024)
+        .with_remote_addr("127.0.0.1:9418");
+
+    assert_eq!(req.method, "GET");
+    assert_eq!(req.path, "/info/refs");
+    assert_eq!(req.query.as_deref(), Some("service=git-upload-pack"));
+    assert_eq!(
+        req.headers.get("Content-Type").map(String::as_str),
+        Some("application/x-git-upload-pack-advertisement")
+    );
+    assert_eq!(req.output_cap, 1024);
+    assert_eq!(req.remote_addr.as_deref(), Some("127.0.0.1:9418"));
+    assert_eq!(req.endpoint, EndpointKind::InfoRefs);
+    assert_eq!(req.service, "test-service");
+}
+
+#[test]
+fn stream_outcome_bytes_round_trip() {
+    assert_eq!(StreamOutcome::Eof.bytes(), None);
+    assert_eq!(StreamOutcome::Bytes(7).bytes(), Some(7));
+}
+
+#[test]
+fn pack_stats_default_is_empty() {
+    let s = PackStats::default();
+    assert!(s.is_empty());
+    assert_eq!(s.bytes_streamed, 0);
+    assert_eq!(s.objects_reported, 0);
+}
+
+#[test]
+fn ingest_report_empty_is_default() {
+    let r = IngestReport::empty();
+    assert!(r.refs_updated.is_empty());
+    assert!(r.stats.is_empty());
+}
+
+#[test]
+fn empty_readers_and_helpers_are_present() {
+    let r1 = empty_reader();
+    let r2 = buf_empty_reader();
+    let rt = tokio::runtime::Builder::new_current_thread()
+        .enable_all()
+        .build()
+        .expect("rt");
+    let n1 = rt.block_on(async {
+        let mut r = r1;
+        let mut buf = [0u8; 4];
+        r.read(&mut buf).await.unwrap_or(0)
+    });
+    let n2 = rt.block_on(async {
+        let mut r = r2;
+        let mut buf = [0u8; 4];
+        r.read(&mut buf).await.unwrap_or(0)
+    });
+    assert_eq!(n1, 0);
+    assert_eq!(n2, 0);
+}
+
+#[tokio::test]
+async fn backend_placeholders_report_unimplemented() {
+    let backend = PlaceholderBackend;
+    let req = make_request("GET", "/info/refs", EndpointKind::InfoRefs);
+    let err = backend
+        .authorize(&req)
+        .await
+        .expect_err("placeholder must return Err");
+    match err {
+        ServerError::Unimplemented(msg) => assert!(msg.contains("authorize")),
+        other => panic!("unexpected error variant: {:?}", other),
+    }
+}
+
+#[tokio::test]
+async fn child_pack_reader_reads_bytes_then_reports_eof() {
+    use std::io::Cursor;
+    let data: Vec<u8> = (0u8..16).collect();
+    let reader = ChildPackReader::new(tokio::io::BufReader::new(Cursor::new(data.clone())), None);
+
+    let mut total: u64 = 0;
+    let mut reader = reader;
+    loop {
+        let mut buf = [0u8; 4];
+        let n = reader.next_chunk(&mut buf).await.expect("read ok");
+        if n == 0 {
+            break;
+        }
+        total = total.saturating_add(n as u64);
+    }
+    assert_eq!(total, 16);
+    assert!(reader.finished());
+}
+
+#[tokio::test]
+async fn child_pack_reader_collects_pack_stats_and_enforces_cap() {
+    use std::io::Cursor;
+    let data: Vec<u8> = (0u8..32).collect();
+    let mut reader = ChildPackReader::new(tokio::io::BufReader::new(Cursor::new(data)), None);
+    let stats = reader
+        .collect_pack_stats(Some(64))
+        .await
+        .expect("within cap");
+    assert_eq!(stats.bytes_streamed, 32);
+
+    let data: Vec<u8> = (0u8..32).collect();
+    let mut reader = ChildPackReader::new(tokio::io::BufReader::new(Cursor::new(data)), None);
+    let err = reader
+        .collect_pack_stats(Some(8))
+        .await
+        .expect_err("cap should reject");
+    match err {
+        ServerError::PayloadTooLarge { .. } => {}
+        other => panic!("unexpected error: {:?}", other),
+    }
+}
+
+#[tokio::test]
+async fn child_pack_writer_copies_reader_to_writer_and_finishes() {
+    use std::io::Cursor;
+    use tokio::io::BufWriter;
+
+    let src: Vec<u8> = (0u8..8).collect();
+    let mut src_reader = tokio::io::BufReader::new(Cursor::new(src));
+    let dst: Vec<u8> = Vec::new();
+    let buf_dst = BufWriter::new(Cursor::new(dst));
+
+    let mut writer = ChildPackWriter::new(buf_dst, None);
+    let outcome = writer.copy_from(&mut src_reader).await.expect("copy ok");
+    match outcome {
+        StreamOutcome::Bytes(n) => assert_eq!(n, 8),
+        StreamOutcome::Eof => panic!("expected bytes outcome"),
+    }
+    writer.finish().await.expect("finish ok");
+}
+
+#[tokio::test]
+async fn server_request_body_streams_bytes() {
+    use std::io::Cursor;
+    let body_data: Vec<u8> = b"PACK".to_vec();
+    let req = ServerRequest::new(
+        "POST",
+        "/git-receive-pack",
+        EndpointKind::GitReceivePack,
+        "test",
+        Box::new(Cursor::new(body_data.clone())),
+    );
+
+    let mut body = req.body;
+    let mut got = Vec::new();
+    body.read_to_end(&mut got).await.expect("read body");
+    assert_eq!(got, body_data);
+}
+
+#[test]
+fn ref_update_entry_serialises_oid_strings_not_objectid() {
+    let entry = RefUpdateEntry {
+        name: "refs/heads/x".into(),
+        new_oid: "deadbeef".into(),
+        old_oid: Some("feedface".into()),
+    };
+    assert_eq!(entry.new_oid, "deadbeef");
+    assert_eq!(entry.old_oid.as_deref(), Some("feedface"));
+}
+
+#[test]
+fn endpoint_kind_str_matches_module_taxonomy() {
+    let _guard = lock_or_recover();
+    assert_eq!(EndpointKind::GitUploadPack.as_str(), "git-upload-pack");
+    assert_eq!(EndpointKind::GitReceivePack.as_str(), "git-receive-pack");
+    assert_eq!(EndpointKind::LfsBatch.as_str(), "lfs-batch");
+    assert_eq!(EndpointKind::LfsObjects.as_str(), "lfs-objects");
+    assert_eq!(EndpointKind::InfoRefs.as_str(), "info-refs");
+    assert_eq!(EndpointKind::SmartHttp.as_str(), "smart-http");
+    assert_eq!(EndpointKind::Custom.as_str(), "custom");
+}
+
+#[test]
+fn guard_clone_is_independent_until_cancel() {
+    let g = StreamGuard::new();
+    let g2 = g.clone();
+    g.cancel();
+    assert!(g.is_cancelled());
+    assert!(g2.is_cancelled());
+}
+
+#[test]
+fn auth_context_default_is_anonymous() {
+    let ctx = AuthContext::default();
+    assert_eq!(ctx.scheme, AuthScheme::Anonymous);
+    assert!(ctx.principal.is_none());
+}
+
+#[test]
+fn server_request_default_output_cap_matches_command_cap() {
+    let req = make_request("GET", "/", EndpointKind::Custom);
+    assert_eq!(
+        req.output_cap,
+        gitbaby::server::DEFAULT_OUTPUT_CAP_BYTES as u64
+    );
+}
+
+#[test]
+fn ref_update_batch_default_is_empty() {
+    let batch = RefUpdateBatch::default();
+    assert!(batch.is_empty());
+    assert_eq!(batch.len(), 0);
+}
+
+#[test]
+fn endpoint_kind_equality_works() {
+    assert_eq!(EndpointKind::InfoRefs, EndpointKind::InfoRefs);
+    assert_ne!(EndpointKind::InfoRefs, EndpointKind::SmartHttp);
+}
+
+#[tokio::test]
+async fn stream_guard_cancel_during_async_does_not_panic() {
+    let guard = Arc::new(StreamGuard::new());
+    let g2 = guard.clone();
+
+    let h = tokio::spawn(async move {
+        tokio::time::sleep(Duration::from_millis(200)).await;
+        g2.cancel();
+    });
+
+    tokio::time::sleep(Duration::from_millis(50)).await;
+    assert!(!guard.is_cancelled());
+
+    h.await.expect("task ok");
+    assert!(guard.is_cancelled());
+}