diff --git a/src/ingestion/connectors/git/bitbucket-cloud/dbt/schema.yml b/src/ingestion/connectors/git/bitbucket-cloud/dbt/schema.yml
index 52e6599fc..753ff15ce 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/dbt/schema.yml
+++ b/src/ingestion/connectors/git/bitbucket-cloud/dbt/schema.yml
@@ -19,7 +19,6 @@ sources:
       - name: pull_requests
       - name: pull_request_diffstat
       - name: pull_request_activity
-      - name: pull_request_tasks
       - name: pull_request_comments
       - name: pull_request_commits
       # Optional streams: permission- or feature-gated endpoints (Pipelines,
@@ -45,6 +44,10 @@ sources:
         freshness: null
       - name: issue_changes
         freshness: null
+      # Not wired into the source right now (no model reads it), so its table
+      # has no newest row to be fresh.
+      - name: pull_request_tasks
+        freshness: null
 
 models:
   - name: bitbucket_cloud__repositories
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/descriptor.yaml b/src/ingestion/connectors/git/bitbucket-cloud/descriptor.yaml
index 3aef6f87f..cbaf69599 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/descriptor.yaml
+++ b/src/ingestion/connectors/git/bitbucket-cloud/descriptor.yaml
@@ -1,9 +1,9 @@
 name: bitbucket-cloud
 version: "1.12.0"
 type: cdk
-# 05:00 — dedicated slot: the 02:00 slot is shared with cursor + m365 and the
-# single-worker node starves replication pods there (Broken-pipe startup
-# deaths, Temporal activity timeouts — jobs 432/454 on 2026-07-15/16).
+# A slot of its own: replication pods sharing a slot with other connector
+# syncs can be starved of workers, which surfaces as startup and activity
+# timeouts rather than as a connector error.
 schedule: "0 5 * * *"
 workflow: sync
 # CI build metadata + runtime image refs (ADR-0016).
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py
index 7acdc56f4..d6a1fc597 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/client.py
@@ -1,7 +1,9 @@
 from __future__ import annotations
 
+import json
 import os
 import random
+import threading
 import time
 from collections.abc import Collection, Iterable, Mapping, Sequence
 from dataclasses import dataclass
@@ -14,6 +16,17 @@
 from source_bitbucket_cloud.auth import auth_headers
 
 
+# Statuses that mean "this token will never read this resource": no retry helps,
+# so the repository is skipped instead of failing the sync. 404 is here too — a
+# repository listed at the start of a sync can be deleted mid-run.
+DENIED_STATUSES = frozenset({403, 404})
+
+# Bitbucket answers 400 for a pull request whose source and destination share
+# no ancestry: the diff is undefined rather than empty, and no retry or later
+# sync can make it computable.
+UNCOMPUTABLE_DIFF = frozenset({"No common ancestor"})
+
+
 class BitbucketApiError(RuntimeError):
     def __init__(self, status_code: int, url: str, body: str) -> None:
         super().__init__(f"Bitbucket API returned {status_code} for {url}: {body[:500]}")
@@ -21,6 +34,30 @@ def __init__(self, status_code: int, url: str, body: str) -> None:
         self.url = url
         self.body = body
 
+    @property
+    def _payload(self) -> Mapping[str, Any]:
+        try:
+            payload = json.loads(self.body)
+        except (TypeError, ValueError):
+            return {}
+        return payload if isinstance(payload, Mapping) else {}
+
+    @property
+    def error_message(self) -> str:
+        error = self._payload.get("error")
+        if not isinstance(error, Mapping):
+            return ""
+        return str(error.get("message") or "")
+
+    @property
+    def missing_shas(self) -> frozenset[str]:
+        error = self._payload.get("error")
+        data = error.get("data") if isinstance(error, Mapping) else None
+        shas = data.get("shas") if isinstance(data, Mapping) else None
+        if not isinstance(shas, list):
+            return frozenset()
+        return frozenset(str(sha) for sha in shas if sha)
+
 
 @dataclass(frozen=True)
 class RepositoryRef:
@@ -33,13 +70,14 @@ class RepositoryRef:
     raw: Mapping[str, Any]
 
 
-@dataclass(frozen=True)
+# Held for every branch of every repository for the length of a sync, so it
+# carries the four fields the streams read and not the API object.
+@dataclass(frozen=True, slots=True)
 class BranchRef:
     name: str
     head_sha: str
     target_date: str | None
     is_default: bool
-    raw: Mapping[str, Any]
 
 
 class RepositoryCatalog:
@@ -50,6 +88,10 @@ def __init__(self, client: BitbucketClient, workspaces: Sequence[str], skip_fork
         self._repositories: list[RepositoryRef] | None = None
         self._branches: dict[str, list[BranchRef]] = {}
         self._inaccessible: set[str] = set()
+        # Repositories are read concurrently, so the memoised fills are guarded;
+        # the selection caches below are keyed per repository and only ever
+        # written by the worker that owns that repository.
+        self._lock = threading.Lock()
         # Shared per-sync selection caches (see streams/pr_base.py). Keyed by
         # (repository, watermark) and holding SLIM projections only — a handful
         # of scalar fields per entity, never the raw API objects. The raw list
@@ -67,36 +109,67 @@ def mark_inaccessible(self, repo: RepositoryRef) -> None:
         permissions. The catalog is shared by every stream, so the first stream
         to discover it saves the others from rediscovering it repo by repo.
         """
-        self._inaccessible.add(repo.uuid)
+        with self._lock:
+            self._inaccessible.add(repo.uuid)
 
     def is_inaccessible(self, repo: RepositoryRef) -> bool:
-        return repo.uuid in self._inaccessible
+        with self._lock:
+            return repo.uuid in self._inaccessible
 
     @property
     def inaccessible_count(self) -> int:
-        return len(self._inaccessible)
+        with self._lock:
+            return len(self._inaccessible)
+
+    @property
+    def branch_cache_size(self) -> tuple[int, int]:
+        with self._lock:
+            return len(self._branches), sum(len(branches) for branches in self._branches.values())
 
     def repositories(self) -> list[RepositoryRef]:
-        if self._repositories is None:
-            self._repositories = self._client.repositories(self._workspaces, self._skip_forks)
-        return self._repositories
+        with self._lock:
+            if self._repositories is not None:
+                return self._repositories
+        fetched = self._client.repositories(self._workspaces, self._skip_forks)
+        with self._lock:
+            if self._repositories is None:
+                self._repositories = fetched
+            return self._repositories
 
     def branches(self, repo: RepositoryRef) -> list[BranchRef]:
-        if repo.uuid not in self._branches:
-            self._branches[repo.uuid] = self._client.branches(repo)
-        return self._branches[repo.uuid]
+        with self._lock:
+            cached = self._branches.get(repo.uuid)
+        if cached is not None:
+            return cached
+        fetched = self._client.branches(repo)
+        with self._lock:
+            return self._branches.setdefault(repo.uuid, fetched)
 
 
 class BitbucketClient:
     url_base = "https://api.bitbucket.org/2.0/"
 
     def __init__(self, token: str, username: str = "", base_url: str | None = None) -> None:
-        self._session = requests.Session()
-        self._session.headers.update(auth_headers(token, username))
-        self._session.headers.update({"Accept": "application/json"})
+        self._headers = {**auth_headers(token, username), "Accept": "application/json"}
+        self._local = threading.local()
         configured_url = base_url or os.environ.get("BITBUCKET_API_BASE_URL") or self.url_base
         self._base_url = configured_url.rstrip("/") + "/"
 
+    @property
+    def _session(self) -> requests.Session:
+        # requests.Session is not thread-safe; repositories are read in
+        # parallel, so each worker gets its own connection pool.
+        session = getattr(self._local, "session", None)
+        if session is None:
+            session = requests.Session()
+            session.headers.update(self._headers)
+            self._local.session = session
+        return session
+
+    @_session.setter
+    def _session(self, session: requests.Session) -> None:
+        self._local.session = session
+
     def request(
         self,
         method: str,
@@ -170,10 +243,50 @@ def paginate(
                 raise ValueError(f"Unexpected Bitbucket response from {response.url}")
             first = False
 
+    def _optional_request(
+        self,
+        path_or_url: str,
+        *,
+        params: Mapping[str, Any] | Sequence[tuple[str, Any]] | None = None,
+        tolerate_messages: Collection[str] = (),
+    ) -> requests.Response | None:
+        try:
+            return self.request("GET", path_or_url, params=params, allow_statuses={403, 404})
+        except BitbucketApiError as error:
+            if error.status_code == 400 and error.error_message in tolerate_messages:
+                return None
+            raise
+
+    def _next_page(self, next_value: Any) -> requests.Response | None:
+        """Fetch a continuation page, refusing to end a collection quietly.
+
+        Tolerating a refusal here would hand the caller part of a collection to
+        publish as a complete snapshot, which deletes whatever the unread pages
+        held. It is not a denial either — the collection was readable a moment
+        ago — so it must not mark the whole repository inaccessible.
+        """
+        if not next_value:
+            return None
+        try:
+            return self.request("GET", str(next_value))
+        except BitbucketApiError as error:
+            if error.status_code not in DENIED_STATUSES:
+                # 401 in particular has to reach the sync untouched: it aborts
+                # the whole read with the cause instead of quarantining every
+                # remaining repository one at a time.
+                raise
+            raise RuntimeError(
+                f"Bitbucket refused a continuation page after the collection had started: {next_value}"
+            ) from error
+
     def paginate_optional(
-        self, path: str, *, params: Mapping[str, Any] | Sequence[tuple[str, Any]] | None = None
+        self,
+        path: str,
+        *,
+        params: Mapping[str, Any] | Sequence[tuple[str, Any]] | None = None,
+        tolerate_messages: Collection[str] = (),
     ) -> tuple[bool, Iterable[Mapping[str, Any]]]:
-        response = self.request("GET", path, params=params, allow_statuses={403, 404})
+        response = self._optional_request(path, params=params, tolerate_messages=tolerate_messages)
         if response is None:
             return False, ()
 
@@ -196,11 +309,7 @@ def records() -> Iterable[Mapping[str, Any]]:
                     raise RuntimeError(f"Bitbucket pagination loop detected for {next_value}")
                 if next_value:
                     seen.add(str(next_value))
-                current = (
-                    self.request("GET", str(next_value), allow_statuses={403, 404})
-                    if next_value
-                    else None
-                )
+                current = self._next_page(next_value)
 
         return True, records()
 
@@ -254,7 +363,6 @@ def branches(self, repo: RepositoryRef) -> list[BranchRef]:
                         head_sha=head,
                         target_date=target.get("date"),
                         is_default=name == repo.mainbranch_name,
-                        raw=raw,
                     )
                 )
         return branches
@@ -267,6 +375,30 @@ def branches(self, repo: RepositoryRef) -> list[BranchRef]:
     # dedups any overlap by unique_key.
     COMMITS_INCLUDE_CHUNK = 100
 
+    # Everything the commit streams read. The default payload additionally
+    # carries the message rendered to HTML, a summary rendering of it again,
+    # and a links map per commit — several times this projection, multiplied by
+    # the highest-volume endpoint in the connector. A field misspelled here is
+    # silently dropped by the API and surfaces as a NULL column, so the list is
+    # pinned by a test against the commits schema.
+    COMMIT_FIELDS = ",".join(
+        [
+            "values.hash",
+            "values.date",
+            "values.message",
+            "values.author.raw",
+            "values.author.user.display_name",
+            "values.author.user.uuid",
+            "values.author.user.account_id",
+            "values.committer.raw",
+            "values.committer.user.display_name",
+            "values.committer.user.uuid",
+            "values.committer.user.account_id",
+            "values.parents.hash",
+            "next",
+        ]
+    )
+
     def commits_between(
         self, repo: RepositoryRef, current_heads: Sequence[str], previous_heads: Sequence[str]
     ) -> Iterable[Mapping[str, Any]]:
@@ -282,7 +414,7 @@ def commits_between(
             yield from self.paginate(
                 self.repo_path(repo, "commits"),
                 method="POST",
-                params={"pagelen": "100"},
+                params={"pagelen": "100", "fields": self.COMMIT_FIELDS},
                 data=form,
             )
 
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/source.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/source.py
index bba0f03a5..298af0b2f 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/source.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/source.py
@@ -11,22 +11,24 @@
 from airbyte_cdk.sources.streams import Stream
 
 from source_bitbucket_cloud.client import BitbucketApiError, BitbucketClient, RepositoryCatalog
+from source_bitbucket_cloud.streams.base import DEFAULT_CONCURRENCY
 from source_bitbucket_cloud.streams.branches import BranchesStream
 from source_bitbucket_cloud.streams.commit_branch_reachability import CommitBranchReachabilityStream
 from source_bitbucket_cloud.streams.commits import CommitsStream
 from source_bitbucket_cloud.streams.file_changes import FileChangesStream
-from source_bitbucket_cloud.streams.metric_events import (
-    DeploymentsStream,
-    EnvironmentsStream,
-    IssueChangesStream,
-    IssueCommentsStream,
-    IssuesStream,
-    PipelinesStream,
-    PipelineStepsStream,
-    PipelineStepTestReportsStream,
-    PRTasksStream,
-    TagsStream,
-)
+# Unwired until a model reads them — see the note in streams().
+# from source_bitbucket_cloud.streams.metric_events import (
+#     DeploymentsStream,
+#     EnvironmentsStream,
+#     IssueChangesStream,
+#     IssueCommentsStream,
+#     IssuesStream,
+#     PipelinesStream,
+#     PipelineStepsStream,
+#     PipelineStepTestReportsStream,
+#     PRTasksStream,
+#     TagsStream,
+# )
 from source_bitbucket_cloud.streams.pr_activity import PRActivityStream
 from source_bitbucket_cloud.streams.pr_comments import PRCommentsStream
 from source_bitbucket_cloud.streams.pr_commits import PRCommitsStream
@@ -102,6 +104,7 @@ def streams(self, config: Mapping[str, Any]) -> list[Stream]:
             "workspaces": config["bitbucket_workspaces"],
             "skip_forks": config.get("bitbucket_skip_forks", True),
             "start_date": config.get("bitbucket_start_date"),
+            "concurrency": int(config.get("bitbucket_concurrency") or DEFAULT_CONCURRENCY),
             "client": client,
             "catalog": catalog,
         }
@@ -116,42 +119,48 @@ def streams(self, config: Mapping[str, Any]) -> list[Stream]:
         pr_activity = PRActivityStream(**shared)
         pr_comments = PRCommentsStream(**shared)
         pr_commits = PRCommitsStream(**shared)
-        pipelines = PipelinesStream(**shared)
-        pipeline_steps = PipelineStepsStream(**shared)
-        pipeline_step_test_reports = PipelineStepTestReportsStream(**shared)
-        deployments = DeploymentsStream(**shared)
-        environments = EnvironmentsStream(**shared)
-        tags = TagsStream(**shared)
-        issues = IssuesStream(**shared)
-        issue_comments = IssueCommentsStream(**shared)
-        issue_changes = IssueChangesStream(**shared)
-        pr_tasks = PRTasksStream(**shared)
 
+        # Airbyte reads streams in catalog order, so a sync that is cut short
+        # keeps whatever the transform layer consumes. Within that, the streams
+        # whose cost a watermark bounds run before the commit-range streams,
+        # whose first read of a repository is bounded only by its history: a run
+        # that ends early then still delivers the bounded half whole.
+        #
+        # Pipelines, tags, deployments, environments, issues and pull-request
+        # tasks are wired out below rather than merely ordered last: no model
+        # reads them, and collecting them took most of a full pass. Re-enabling
+        # one means uncommenting it and its import — the stream classes, schemas
+        # and tests are all still here — but two things come with it. The
+        # platform may have dropped the state of a stream it no longer sees, so
+        # the first pass back can be a backfill; and tags pages a repository's
+        # whole tag history every time, so it wants the idle gate branches uses
+        # before it returns to a large workspace.
+        # pipelines = PipelinesStream(**shared)
+        # pipeline_steps = PipelineStepsStream(**shared)
+        # pipeline_step_test_reports = PipelineStepTestReportsStream(**shared)
+        # deployments = DeploymentsStream(**shared)
+        # environments = EnvironmentsStream(**shared)
+        # tags = TagsStream(**shared)
+        # issues = IssuesStream(**shared)
+        # issue_comments = IssueCommentsStream(**shared)
+        # issue_changes = IssueChangesStream(**shared)
+        # pr_tasks = PRTasksStream(**shared)
         streams = [
             repos,
             branches,
             prs,
-            pr_diffstat,
-            pr_activity,
-            pr_tasks,
-            pr_comments,
             pr_commits,
-            pipelines,
-            pipeline_steps,
-            pipeline_step_test_reports,
-            deployments,
-            environments,
-            tags,
-            issues,
-            issue_comments,
-            issue_changes,
+            pr_comments,
+            pr_activity,
+            pr_diffstat,
             commits,
-            commit_branch_reachability,
             file_changes,
+            commit_branch_reachability,
         ]
         _logger.info(
             f"streams: wired {len(streams)} streams (workspaces={shared['workspaces']} "
-            f"start_date={shared['start_date']} skip_forks={shared['skip_forks']})"
+            f"start_date={shared['start_date']} skip_forks={shared['skip_forks']} "
+            f"concurrency={shared['concurrency']})"
         )
         return streams
 
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/spec.json b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/spec.json
index e1e06712f..b3fa983c6 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/spec.json
+++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/spec.json
@@ -65,6 +65,15 @@
         "format": "uri",
         "default": "https://api.bitbucket.org/2.0/",
         "order": 7
+      },
+      "bitbucket_concurrency": {
+        "type": "integer",
+        "title": "Concurrent Repositories",
+        "description": "How many repositories to read at once. Raise it for a large workspace; lower it to 1 if the API starts rate limiting.",
+        "default": 4,
+        "minimum": 1,
+        "maximum": 16,
+        "order": 8
       }
     }
   }
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py
index fb19c98c8..bee093ae0 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/base.py
@@ -3,17 +3,22 @@
 import hashlib
 import json
 import logging
+import queue
 import re
+import threading
 import uuid
 from abc import ABC
+from collections import deque
 from collections.abc import Iterable, Mapping, MutableMapping, Sequence
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime, timezone
-from typing import Any
+from typing import Any, NamedTuple
 
 from airbyte_cdk.models import SyncMode
 from airbyte_cdk.sources.streams import CheckpointMixin, Stream
 
 from source_bitbucket_cloud.client import (
+    DENIED_STATUSES,
     BitbucketApiError,
     BitbucketClient,
     RepositoryCatalog,
@@ -23,14 +28,41 @@
 logger = logging.getLogger("airbyte")
 
 BUCKET_COUNT = 8
-MAX_TEXT_BYTES = 16_384
+# Commit messages, PR descriptions and comment bodies are kept for display,
+# not parsed: nothing downstream reads past the opening lines, while generated
+# descriptions routinely run to tens of KB and multiply bronze storage.
+MAX_TEXT_BYTES = 2_048
 # Bumped from 2 when entity and state keys moved back to workspace/slug: a
 # version-2 state is keyed by repository uuid and no longer addresses anything.
 STATE_VERSION = 3
-# Statuses that mean "this token will never read this repository": no retry
-# helps, so the repository is skipped instead of failing the sync. 404 is here
-# too — a repository listed at the start of a sync can be deleted mid-run.
-DENIED_STATUSES = frozenset({403, 404})
+# Records a worker may run ahead of the consumer, per repository. Bounded so a
+# repository with a long history cannot buffer itself into memory.
+RECORD_BUFFER = 500
+# Records taken from one repository before moving to the next. A producer that
+# refills faster than the consumer emits would otherwise hold the floor until
+# its repository finished, leaving every other worker parked on a full buffer.
+DRAIN_BATCH = 64
+QUEUE_POLL_SECONDS = 0.5
+DEFAULT_CONCURRENCY = 4
+MAX_CONCURRENCY = 16
+_READ_DONE = object()
+
+
+class _PendingRead(NamedTuple):
+    repo: RepositoryRef
+    records: queue.Queue[Any]
+
+
+class _StagedState(NamedTuple):
+    """State a worker produced, travelling behind that worker's records.
+
+    A checkpoint may be taken between any two records the consumer emits, so
+    state must only claim a repository once its records have actually left —
+    otherwise a crash in that window loses them and the idle gate skips the
+    repository on the next sync.
+    """
+
+    entries: Mapping[str, Mapping[str, Any]]
 
 
 def now_iso() -> str:
@@ -161,10 +193,12 @@ def __init__(
         username: str = "",
         skip_forks: bool = True,
         start_date: str | None = None,
+        concurrency: int = 1,
         client: BitbucketClient | None = None,
         catalog: RepositoryCatalog | None = None,
     ) -> None:
         self._client = client or BitbucketClient(token, username)
+        self._concurrency = max(1, min(MAX_CONCURRENCY, concurrency))
         self._tenant_id = tenant_id
         self._source_id = source_id
         self._workspaces = tuple(workspaces)
@@ -197,46 +231,201 @@ def read_records(
     ) -> Iterable[Mapping[str, Any]]:
         del sync_mode, cursor_field, stream_state
         bucket_id, repositories = self.bucket(stream_slice)
+        read = self._read_serially if self._concurrency <= 1 else self._read_concurrently
+        yield from read(bucket_id, repositories)
+        self.finish_bucket(bucket_id, repositories)
+
+    def _read_serially(
+        self, bucket_id: int, repositories: Sequence[RepositoryRef]
+    ) -> Iterable[Mapping[str, Any]]:
         for repo in repositories:
-            if self._catalog.is_inaccessible(repo):
-                # Discovered by an earlier stream; still counts toward THIS
-                # stream's end-of-sync skipped summary.
-                self._skipped_repositories.append(f"{repo.workspace}/{repo.slug}")
+            if self.already_denied(repo):
                 continue
             try:
                 yield from self.repository_records(repo, bucket_id)
-            except BitbucketApiError as error:
-                if error.status_code == 401:
-                    # Credential failure is global, not per-repository: every
-                    # remaining repo would fail identically, drowning the log in
-                    # quarantine noise before a generic end-of-sync error. Abort
-                    # now with the actionable cause instead.
-                    raise RuntimeError(
-                        "Bitbucket authentication failed mid-sync (HTTP 401): the token was "
-                        "rejected. If bitbucket_username is unset, Atlassian API tokens are "
-                        "sent as Bearer and refused — set the username, or the token has "
-                        "expired/been rotated."
-                    ) from error
-                if error.status_code in DENIED_STATUSES:
-                    self.skip_repository(repo, error.status_code)
-                else:
-                    self.record_failure(repo)
-            except Exception:
-                self.record_failure(repo)
-        self.finish_bucket(bucket_id, repositories)
+            except BaseException as error:  # noqa: BLE001 - classified below
+                self.handle_repository_error(repo, error)
+
+    def _read_concurrently(
+        self, bucket_id: int, repositories: Sequence[RepositoryRef]
+    ) -> Iterable[Mapping[str, Any]]:
+        """Read several repositories at once, emit whatever is ready.
+
+        Fetching is the whole cost of a bucket and the per-repository reads are
+        independent. Consuming in submission order would let one repository
+        with a deep history park every other worker on a full buffer and block
+        new submissions behind it; records interleave across repositories
+        instead, which bronze does not mind (append-only, keyed). Failures stay
+        attributed: each repository has its own queue, and its error travels on
+        it.
+        """
+        stop = threading.Event()
+        pending: deque[_PendingRead] = deque()
+        waiting = iter(repositories)
+        with ThreadPoolExecutor(max_workers=self._concurrency) as pool:
+            # Inside the pool: leaving the sync early (the consumer stops
+            # reading, or a worker raises) must release the workers parked on a
+            # full buffer before shutdown waits for them.
+            try:
+                while True:
+                    while len(pending) < self._concurrency:
+                        repo = next(waiting, None)
+                        if repo is None:
+                            break
+                        if self.already_denied(repo):
+                            continue
+                        pending.append(self._submit(pool, repo, bucket_id, stop))
+                    if not pending:
+                        return
+                    yield from self._drain_ready(pending)
+            finally:
+                stop.set()
+
+    def _drain_ready(self, pending: deque[_PendingRead]) -> Iterable[Mapping[str, Any]]:
+        drained_any = False
+        for read in list(pending):
+            for _ in range(DRAIN_BATCH):
+                try:
+                    item = read.records.get_nowait()
+                except queue.Empty:
+                    break
+                drained_any = True
+                if item is _READ_DONE:
+                    pending.remove(read)
+                    break
+                if isinstance(item, _StagedState):
+                    self.apply_staged_state(item.entries)
+                    continue
+                if isinstance(item, BaseException):
+                    self.handle_repository_error(read.repo, item)
+                    continue
+                yield item
+
+        if drained_any or not pending:
+            return
+        # Nothing ready anywhere: block briefly on the oldest read rather than
+        # spinning; the sweep resumes with whatever else arrived meanwhile.
+        oldest = pending[0]
+        try:
+            item = oldest.records.get(timeout=QUEUE_POLL_SECONDS)
+        except queue.Empty:
+            return
+        if item is _READ_DONE:
+            pending.remove(oldest)
+        elif isinstance(item, _StagedState):
+            self.apply_staged_state(item.entries)
+        elif isinstance(item, BaseException):
+            self.handle_repository_error(oldest.repo, item)
+        else:
+            yield item
+
+    def _submit(
+        self, pool: ThreadPoolExecutor, repo: RepositoryRef, bucket_id: int, stop: threading.Event
+    ) -> _PendingRead:
+        records: queue.Queue[Any] = queue.Queue(maxsize=RECORD_BUFFER)
+        pool.submit(self._collect, repo, bucket_id, records, stop)
+        return _PendingRead(repo, records)
+
+    def _collect(
+        self, repo: RepositoryRef, bucket_id: int, records: queue.Queue[Any], stop: threading.Event
+    ) -> None:
+        staged = self.stage_state()
+        try:
+            for record in self.repository_records(repo, bucket_id):
+                if not self._offer(records, record, stop):
+                    return
+        except BaseException as error:  # noqa: BLE001 - re-raised in the consumer
+            self._offer(records, error, stop)
+        finally:
+            self.stop_staging()
+            if staged:
+                self._offer(records, _StagedState(staged), stop)
+            self._offer(records, _READ_DONE, stop)
+
+    def stage_state(self) -> MutableMapping[str, Mapping[str, Any]]:
+        """Redirect this worker's state commits into a buffer it owns."""
+        return {}
+
+    def stop_staging(self) -> None:
+        return
+
+    def apply_staged_state(self, entries: Mapping[str, Mapping[str, Any]]) -> None:
+        del entries
+
+    @staticmethod
+    def _offer(records: queue.Queue[Any], item: Any, stop: threading.Event) -> bool:
+        """Park on a full buffer for as long as the read is still wanted.
+
+        No timeout: a full queue cannot tell an absent consumer from a slow
+        destination, and giving up on the second would abandon a repository
+        mid-read. `stop` belongs to the consumer, which sets it once it is
+        finished with the bucket — by which point nothing waits on these
+        records.
+        """
+        while not stop.is_set():
+            try:
+                records.put(item, timeout=QUEUE_POLL_SECONDS)
+            except queue.Full:
+                continue
+            return True
+        return False
+
+    def already_denied(self, repo: RepositoryRef) -> bool:
+        # Discovered by an earlier stream; still counts toward THIS stream's
+        # end-of-sync skipped summary.
+        if not self._catalog.is_inaccessible(repo):
+            return False
+        self._skipped_repositories.append(f"{repo.workspace}/{repo.slug}")
+        return True
+
+    def handle_repository_error(self, repo: RepositoryRef, error: BaseException) -> None:
+        if isinstance(error, BitbucketApiError):
+            if error.status_code == 401:
+                # Credential failure is global, not per-repository: every
+                # remaining repo would fail identically, drowning the log in
+                # quarantine noise before a generic end-of-sync error. Abort
+                # now with the actionable cause instead.
+                raise RuntimeError(
+                    "Bitbucket authentication failed mid-sync (HTTP 401): the token was "
+                    "rejected. If bitbucket_username is unset, Atlassian API tokens are "
+                    "sent as Bearer and refused — set the username, or the token has "
+                    "expired/been rotated."
+                ) from error
+            if error.status_code in DENIED_STATUSES:
+                self.skip_repository(repo, error.status_code)
+                return
+        if not isinstance(error, Exception):
+            raise error
+        self.record_failure(repo, error)
 
     def repository_records(self, repo: RepositoryRef, bucket_id: int) -> Iterable[Mapping[str, Any]]:
         raise NotImplementedError
 
+    def before_start_date(self, timestamp: Any) -> bool:
+        return bool(self._start_date and timestamp and str(timestamp)[:10] < self._start_date)
+
+    def out_of_window(self, repo: RepositoryRef) -> bool:
+        """True when nothing a push produces can fall inside the start window.
+
+        Commits, branches and tags only appear by being pushed, and a push
+        moves the repository's updated_on, so a repository last touched before
+        start_date holds nothing this sync is asked for.
+        """
+        updated_on = str(repo.raw.get("updated_on") or "")
+        return bool(self._start_date and updated_on and updated_on[:10] < self._start_date)
+
     def bucket(self, stream_slice: Mapping[str, Any] | None) -> tuple[int, list[RepositoryRef]]:
         bucket_id = int((stream_slice or {}).get("bucket_id", 0))
         return bucket_id, self.repositories_for_slice(stream_slice)
 
-    def record_failure(self, repo: RepositoryRef) -> None:
+    def record_failure(self, repo: RepositoryRef, error: BaseException | None = None) -> None:
         """A failure worth surfacing: transient, so retrying the sync may fix it."""
         name = f"{repo.workspace}/{repo.slug}"
         self._failed_repositories.append(name)
-        logger.exception(f"{self.name}: repository {name} failed; its state was not advanced, continuing")
+        logger.error(
+            f"{self.name}: repository {name} failed; its state was not advanced, continuing",
+            exc_info=error or True,
+        )
 
     def skip_repository(self, repo: RepositoryRef, status_code: int) -> None:
         """A repository the token cannot read: skip it without failing the sync.
@@ -261,12 +450,18 @@ def skip_repository(self, repo: RepositoryRef, status_code: int) -> None:
 
     def finish_bucket(self, bucket_id: int, repositories: Sequence[RepositoryRef]) -> None:
         del repositories
-        if bucket_id == BUCKET_COUNT - 1 and self._skipped_repositories:
+        if bucket_id != BUCKET_COUNT - 1:
+            return
+
+        cached_repositories, cached_branches = self._catalog.branch_cache_size
+        logger.info(f"{self.name}: branch_cache repositories={cached_repositories} branches={cached_branches}")
+
+        if self._skipped_repositories:
             logger.info(
                 f"{self.name}: skipped {len(self._skipped_repositories)} inaccessible "
                 f"repositories: {', '.join(sorted(set(self._skipped_repositories))[:10])}"
             )
-        if bucket_id == BUCKET_COUNT - 1 and self._failed_repositories:
+        if self._failed_repositories:
             raise RuntimeError(
                 f"{self.name}: {len(self._failed_repositories)} repositories failed this sync: "
                 + ", ".join(self._failed_repositories[:10])
@@ -335,20 +530,38 @@ def generation(self, *parts: Any) -> str:
 
 
 class BitbucketIncrementalStream(BitbucketStream, CheckpointMixin, ABC):
+    # Emit state mid-bucket too: a bucket spans a large share of the fleet, and
+    # a pod that dies between bucket boundaries would otherwise re-read hours
+    # of finished repositories. The interval is coarse because each message
+    # carries the whole repositories map.
+    state_checkpoint_interval = 25_000
+
     def __init__(self, **kwargs: Any) -> None:
         super().__init__(**kwargs)
-        self._state: MutableMapping[str, Any] = {}
+        # Versioned from the start: state emitted without one reads back as
+        # pre-rewrite state and gets reshaped into something addressing nothing.
+        self._state: MutableMapping[str, Any] = self._empty_state()
+        self._state_lock = threading.Lock()
+        self._staging = threading.local()
 
     @property
     def state(self) -> MutableMapping[str, Any]:
-        return self._state
+        # A snapshot, not the live dict: the platform serialises this while
+        # workers commit repositories. Only finished repositories are ever in
+        # the map (each is committed whole), so any snapshot is a valid resume
+        # point.
+        with self._state_lock:
+            return {**self._state, "repositories": dict(self._state.get("repositories") or {})}
 
     @state.setter
     def state(self, value: MutableMapping[str, Any]) -> None:
         if not value:
             self._state = self._empty_state()
-        elif value.get("version") == STATE_VERSION and value.get("bucket_count") == BUCKET_COUNT:
-            self._state = value
+        elif value.get("version") == STATE_VERSION:
+            # bucket_count is not part of the address: keys are repository
+            # scoped and the bucket is derived by hash at read time, so state
+            # written under any bucket count resumes under any other.
+            self._state = {**value, "bucket_count": BUCKET_COUNT}
         elif "version" not in value:
             # Pre-rewrite state: a flat partition -> cursor map. Reshape it so
             # the sync resumes from those checkpoints (see migrate_legacy_state).
@@ -367,18 +580,37 @@ def _empty_state() -> dict[str, Any]:
         return {"version": STATE_VERSION, "bucket_count": BUCKET_COUNT, "repositories": {}}
 
     def repository_state(self, repo: RepositoryRef) -> MutableMapping[str, Any]:
-        repositories = self._state.setdefault("repositories", {})
-        return dict(repositories.get(repo_state_key(repo)) or {})
+        with self._state_lock:
+            repositories = self._state.setdefault("repositories", {})
+            return dict(repositories.get(repo_state_key(repo)) or {})
 
     def commit_repository_state(self, repo: RepositoryRef, value: Mapping[str, Any]) -> None:
-        self._state.setdefault("repositories", {})[repo_state_key(repo)] = dict(value)
+        staged = getattr(self._staging, "entries", None)
+        if staged is not None:
+            staged[repo_state_key(repo)] = dict(value)
+            return
+        with self._state_lock:
+            self._state.setdefault("repositories", {})[repo_state_key(repo)] = dict(value)
+
+    def stage_state(self) -> MutableMapping[str, Mapping[str, Any]]:
+        self._staging.entries = {}
+        return self._staging.entries
+
+    def stop_staging(self) -> None:
+        self._staging.entries = None
+
+    def apply_staged_state(self, entries: Mapping[str, Mapping[str, Any]]) -> None:
+        with self._state_lock:
+            self._state.setdefault("repositories", {}).update(entries)
 
     def prune_bucket_state(self, bucket_id: int, repositories: Sequence[RepositoryRef]) -> None:
         current = {repo_state_key(repo) for repo in repositories}
-        state_repositories = self._state.setdefault("repositories", {})
-        stale = [key for key in state_repositories if repository_bucket(key) == bucket_id and key not in current]
-        for key in stale:
-            del state_repositories[key]
+
+        with self._state_lock:
+            state_repositories = self._state.setdefault("repositories", {})
+            stale = [key for key in state_repositories if repository_bucket(key) == bucket_id and key not in current]
+            for key in stale:
+                del state_repositories[key]
 
     def finish_bucket(self, bucket_id: int, repositories: Sequence[RepositoryRef]) -> None:
         self.prune_bucket_state(bucket_id, repositories)
@@ -386,9 +618,10 @@ def finish_bucket(self, bucket_id: int, repositories: Sequence[RepositoryRef]) -
         super().finish_bucket(bucket_id, repositories)
 
     def log_state_size(self) -> None:
-        encoded = json.dumps(self._state, separators=(",", ":")).encode("utf-8")
+        snapshot = self.state
+        encoded = json.dumps(snapshot, separators=(",", ":")).encode("utf-8")
         logger.info(
-            f"{self.name}: state_repositories={len(self._state.get('repositories', {}))} state_bytes={len(encoded)}"
+            f"{self.name}: state_repositories={len(snapshot.get('repositories', {}))} state_bytes={len(encoded)}"
         )
 
 
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/branches.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/branches.py
index cca0e581f..a3307658a 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/branches.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/branches.py
@@ -13,14 +13,18 @@ class BranchesStream(BitbucketIncrementalStream):
     that is denied (403) or fails simply produces no marker this sync, so dbt
     keeps its previous branch generation, while every other repository updates
     independently. A bucket-scoped generation would freeze branch updates for a
-    whole bucket over one denied repository — and workspaces where unreadable
-    repositories are common (observed in production) would freeze every bucket.
+    whole bucket over one denied repository — and in a workspace where
+    unreadable repositories are common, every bucket would freeze.
 
     Incremental only in the cheapest sense: the per-repository state holds the
     repository's updated_on from the workspace listing, and a repository that
     has not been pushed to since the last pass is skipped without a request —
     its previous generation simply stays the newest complete one.
 
+    start_date does not apply here. Branches are current state, not dated
+    history, so a repository last pushed before the window still has branches
+    worth reporting; the idle gate already keeps it at one listing ever.
+
     Trade-off: a repository deleted from the workspace stops producing
     generations, so its last branch snapshot lingers in silver. That is bounded
     (the repository is gone) and preferable to fleet-wide starvation.
@@ -34,9 +38,22 @@ def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]
         repo_updated_on = str(repo.raw.get("updated_on") or "")
         if repo_updated_on and prior.get("repo_updated_on") == repo_updated_on:
             return
+        branches = self._catalog.branches(repo)
+        if not branches and prior.get("branch_count") != 0:
+            # A snapshot replaces the previous one, so publishing an empty one
+            # deletes every branch this repository had. One empty answer is
+            # never enough to do that — including the first one seen, since
+            # state written before this rule existed carries no count. Record
+            # the observation and hold the cursor; a second consecutive empty
+            # listing is the repository, not the API, and publishes normally.
+            self.commit_repository_state(
+                repo, {"repo_updated_on": str(prior.get("repo_updated_on") or ""), "branch_count": 0}
+            )
+            return
+
         generation = self.generation("branches", *repo_scope(repo))
         entity_keys: set[str] = set()
-        for branch in self._catalog.branches(repo):
+        for branch in branches:
             entity_key = unique_key(self._tenant_id, self._source_id, *repo_scope(repo), branch.name)
             entity_keys.add(entity_key)
             yield self.item(
@@ -65,7 +82,9 @@ def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]
             workspace=repo.workspace,
             repo_slug=repo.slug,
         )
-        self.commit_repository_state(repo, {"repo_updated_on": repo_updated_on})
+        self.commit_repository_state(
+            repo, {"repo_updated_on": repo_updated_on, "branch_count": len(entity_keys)}
+        )
 
     def get_json_schema(self) -> Mapping[str, Any]:
         nullable_string = {"type": ["null", "string"]}
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.py
index e06145e8a..6da6dc435 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commit_branch_reachability.py
@@ -1,19 +1,25 @@
 from __future__ import annotations
 
-from collections.abc import Iterable, Mapping
+from collections.abc import Iterable, Iterator, Mapping
+from itertools import chain
 from typing import Any
 
-from source_bitbucket_cloud.client import BitbucketApiError
+from source_bitbucket_cloud.client import BitbucketApiError, BranchRef, RepositoryRef
 from source_bitbucket_cloud.streams.base import BitbucketIncrementalStream, repo_scope, schema, unique_key
 from source_bitbucket_cloud.streams.git_ranges import CommitRangeMixin
 
 
+RANGE_PREFETCH = 500
+
+
 class CommitBranchReachabilityStream(CommitRangeMixin, BitbucketIncrementalStream):
     name = "commit_branch_reachability"
     cursor_field = "branch_head_sha"
 
     def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]]:
         del bucket_id
+        if self.out_of_window(repo):
+            return
         prior = self.repository_state(repo)
         repo_updated_on = str(repo.raw.get("updated_on") or "")
         if repo_updated_on and prior.get("repo_updated_on") == repo_updated_on:
@@ -25,12 +31,20 @@ def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]
             return
         branches, current_heads = self.branch_snapshot(repo)
         previous_heads = prior.get("heads") or {}
+        if previous_heads and not current_heads:
+            # Every stored branch would read as deleted, and a later listing
+            # that finds them again emits no correction. An answer this
+            # sweeping is not trusted: nothing is emitted, nothing advances.
+            return
         branch_by_name = {branch.name: branch for branch in branches}
+        unresolved: set[str] = set()
         for branch_name in sorted(set(current_heads) | set(previous_heads)):
             old_head = previous_heads.get(branch_name)
             new_head = current_heads.get(branch_name)
             if old_head == new_head:
                 continue
+            if new_head and not old_head and not self.head_in_window(branch_by_name[branch_name]):
+                continue
             if new_head:
                 yield from self._changes(
                     repo,
@@ -38,6 +52,7 @@ def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]
                     new_head,
                     old_head,
                     "added",
+                    unresolved,
                 )
             if old_head and new_head:
                 yield from self._changes(
@@ -46,6 +61,7 @@ def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]
                     old_head,
                     new_head,
                     "removed",
+                    unresolved,
                 )
             if old_head and not new_head:
                 entity_key = unique_key(
@@ -69,12 +85,50 @@ def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]
                     committed_at=None,
                     reachability_action="branch_deleted",
                 )
-        self.commit_repository_state(repo, {"heads": current_heads, "repo_updated_on": repo_updated_on})
+        stored = {
+            name: head for name, head in self.retained_heads(current_heads, previous_heads).items()
+            if name not in unresolved
+        }
+        complete = self.complete_read(
+            current_heads, unresolved, empty_confirmed=self.empty_listing_confirmed(prior, "heads")
+        )
+        self.commit_repository_state(
+            repo,
+            {
+                "heads": stored,
+                "repo_updated_on": self.cursor_value(prior, repo_updated_on, complete),
+            },
+        )
 
-    def _changes(self, repo, branch, include: str, exclude: str | None, action: str):
+    def _changes(
+        self,
+        repo: RepositoryRef,
+        branch: BranchRef,
+        include: str,
+        exclude: str | None,
+        action: str,
+        unresolved: set[str],
+    ) -> Iterable[Mapping[str, Any]]:
+        # Recovery below has to replace the whole range, so nothing may have
+        # been emitted when it runs — but a first read of a branch spans its
+        # entire history and several repositories are in flight at once. Hold
+        # only the head of the range: the API rejects a stale exclude on the
+        # first request, so a recoverable failure lands inside this window,
+        # while a longer range spills into a plain stream.
+        prefetched: list[Mapping[str, Any]] = []
+        commits: Iterator[Mapping[str, Any]] = iter(())
         try:
-            commits = list(self._client.commits_between(repo, [include], [exclude] if exclude else []))
+            commits = iter(self._client.commits_between(repo, [include], [exclude] if exclude else []))
+            for commit in commits:
+                prefetched.append(commit)
+                if len(prefetched) >= RANGE_PREFETCH:
+                    break
         except BitbucketApiError as exc:
+            if exc.status_code == 404 and include in exc.missing_shas:
+                # The head this range starts from is gone: nothing is reachable
+                # from it, and no other branch of the repository is affected.
+                unresolved.add(branch.name)
+                return
             if exc.status_code != 404 or not exclude:
                 raise
             if action == "added":
@@ -103,12 +157,12 @@ def _changes(self, repo, branch, include: str, exclude: str | None, action: str)
                 reachability_action="removal_unavailable",
             )
             return
-        yield from self._reachability_records(repo, branch, include, action, commits)
+        yield from self._reachability_records(repo, branch, include, action, chain(prefetched, commits))
 
     def _reachability_records(self, repo, branch, head, action, commits):
         for commit in commits:
             committed_at = commit.get("date")
-            if self._start_date and committed_at and str(committed_at)[:10] < self._start_date:
+            if self.before_start_date(committed_at):
                 continue
             sha = str(commit.get("hash") or "")
             if not sha:
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.py
index 909079522..1021e9d00 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/commits.py
@@ -13,6 +13,8 @@ class CommitsStream(CommitRangeMixin, BitbucketIncrementalStream):
 
     def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]]:
         del bucket_id
+        if self.out_of_window(repo):
+            return
         prior = self.repository_state(repo)
         repo_updated_on = str(repo.raw.get("updated_on") or "")
         if repo_updated_on and prior.get("repo_updated_on") == repo_updated_on:
@@ -22,16 +24,29 @@ def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]
             # range fetch entirely. This is what keeps the per-repository
             # request budget at zero for the idle majority of a large fleet.
             return
-        _, current_heads = self.branch_snapshot(repo)
+        branches, current_heads = self.branch_snapshot(repo)
         current_head_shas = sorted(set(current_heads.values()))
         previous_head_shas = prior.get("head_shas") or []
+        unresolved: set[str] = set()
         if current_head_shas != previous_head_shas:
-            for commit in self.new_commits(repo, current_head_shas, previous_head_shas):
+            includes = current_head_shas if previous_head_shas else self.cold_includes(branches)
+            for commit in self.new_commits(repo, includes, previous_head_shas, unresolved):
                 record = self._record(repo, commit)
-                if self._start_date and record.get("date") and str(record["date"])[:10] < self._start_date:
+                if self.before_start_date(record.get("date")):
                     continue
                 yield record
-        self.commit_repository_state(repo, {"head_shas": current_head_shas, "repo_updated_on": repo_updated_on})
+
+        stored = [sha for sha in self.retained_heads(current_head_shas, previous_head_shas) if sha not in unresolved]
+        complete = self.complete_read(
+            current_head_shas, unresolved, empty_confirmed=self.empty_listing_confirmed(prior, "head_shas")
+        )
+        self.commit_repository_state(
+            repo,
+            {
+                "head_shas": stored,
+                "repo_updated_on": self.cursor_value(prior, repo_updated_on, complete),
+            },
+        )
 
     def _record(self, repo, commit: Mapping[str, Any]) -> Mapping[str, Any]:
         sha = str(commit.get("hash") or "")
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.py
index d6e70e552..1641d9a1b 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/file_changes.py
@@ -3,6 +3,7 @@
 from collections.abc import Iterable, Mapping
 from typing import Any
 
+from source_bitbucket_cloud.client import UNCOMPUTABLE_DIFF
 from source_bitbucket_cloud.streams.base import BitbucketIncrementalStream, repo_scope, schema, unique_key
 from source_bitbucket_cloud.streams.git_ranges import CommitRangeMixin
 
@@ -13,6 +14,8 @@ class FileChangesStream(CommitRangeMixin, BitbucketIncrementalStream):
 
     def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]]:
         del bucket_id
+        if self.out_of_window(repo):
+            return
         prior = self.repository_state(repo)
         repo_updated_on = str(repo.raw.get("updated_on") or "")
         if repo_updated_on and prior.get("repo_updated_on") == repo_updated_on:
@@ -22,16 +25,29 @@ def repository_records(self, repo, bucket_id: int) -> Iterable[Mapping[str, Any]
             # range fetch entirely. This is what keeps the per-repository
             # request budget at zero for the idle majority of a large fleet.
             return
-        _, current_heads = self.branch_snapshot(repo)
+        branches, current_heads = self.branch_snapshot(repo)
         current_head_shas = sorted(set(current_heads.values()))
         previous_head_shas = prior.get("head_shas") or []
+        unresolved: set[str] = set()
         if current_head_shas != previous_head_shas:
-            for commit in self.new_commits(repo, current_head_shas, previous_head_shas):
+            includes = current_head_shas if previous_head_shas else self.cold_includes(branches)
+            for commit in self.new_commits(repo, includes, previous_head_shas, unresolved):
                 committed_date = commit.get("date")
-                if self._start_date and committed_date and str(committed_date)[:10] < self._start_date:
+                if self.before_start_date(committed_date):
                     continue
                 yield from self._diffstat(repo, str(commit.get("hash") or ""), committed_date)
-        self.commit_repository_state(repo, {"head_shas": current_head_shas, "repo_updated_on": repo_updated_on})
+
+        stored = [sha for sha in self.retained_heads(current_head_shas, previous_head_shas) if sha not in unresolved]
+        complete = self.complete_read(
+            current_head_shas, unresolved, empty_confirmed=self.empty_listing_confirmed(prior, "head_shas")
+        )
+        self.commit_repository_state(
+            repo,
+            {
+                "head_shas": stored,
+                "repo_updated_on": self.cursor_value(prior, repo_updated_on, complete),
+            },
+        )
 
     def _diffstat(self, repo, sha: str, committed_date: Any) -> Iterable[Mapping[str, Any]]:
         if not sha:
@@ -45,7 +61,9 @@ def _diffstat(self, repo, sha: str, committed_date: Any) -> Iterable[Mapping[str
         # completeness gate keeps whatever was known before instead of treating
         # the empty read as "this commit changed nothing".
         present, entries = self._client.paginate_optional(
-            self._client.repo_path(repo, f"diffstat/{sha}"), params={"pagelen": "100"}
+            self._client.repo_path(repo, f"diffstat/{sha}"),
+            params={"pagelen": "100"},
+            tolerate_messages=UNCOMPUTABLE_DIFF,
         )
         for entry in entries:
             new_file = entry.get("new") or {}
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.py
index f39d25448..c5db81612 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/git_ranges.py
@@ -1,9 +1,28 @@
 from __future__ import annotations
 
-from collections.abc import Iterable, Mapping, Sequence
+import logging
+from collections.abc import Collection, Iterable, Mapping, Sequence
+from datetime import date, timedelta
+from typing import Any, TypeVar
 
 from source_bitbucket_cloud.client import BitbucketApiError, BranchRef, RepositoryCatalog, RepositoryRef
 
+logger = logging.getLogger("airbyte")
+
+Heads = TypeVar("Heads", list[str], dict[str, str])
+
+# A branch head this far behind the window is not ranged at all: reading it
+# pages the whole history it points at for commits the date filter then throws
+# away. Deliberately lossy past the margin — commit dates are user-supplied, so
+# an ancestor dated inside the window can hang off an older head and is then
+# never collected. The margin is the tolerance for that.
+COLD_START_MARGIN = timedelta(days=90)
+
+# Bitbucket names only the unresolvable shas it noticed, so a repository with
+# several dead heads needs more than one pruning round. Past this many the
+# listing is misbehaving badly enough to say so out loud.
+RANGE_REPAIR_ATTEMPTS = 8
+
 
 class CommitRangeMixin:
     _client: object
@@ -13,15 +32,91 @@ def branch_snapshot(self, repo: RepositoryRef) -> tuple[list[BranchRef], dict[st
         branches = self._catalog.branches(repo)
         return branches, {branch.name: branch.head_sha for branch in branches}
 
+    def head_in_window(self, branch: BranchRef) -> bool:
+        """Whether a never-ranged branch is worth reading from scratch."""
+        floor = self._cold_floor()
+        if floor is None or not branch.target_date:
+            return True
+        return str(branch.target_date)[:10] >= floor
+
+    def cold_includes(self, branches: Sequence[BranchRef]) -> list[str]:
+        return sorted({branch.head_sha for branch in branches if self.head_in_window(branch)})
+
+    def _cold_floor(self) -> str | None:
+        start_date = getattr(self, "_start_date", None)
+        if not start_date:
+            return None
+        return (date.fromisoformat(start_date) - COLD_START_MARGIN).isoformat()
+
+    def retained_heads(self, current: Heads, previous: Heads) -> Heads:
+        """Never trade a known head set for an empty listing.
+
+        Stored heads are only ever the exclude side of the next range, so a
+        stale one can suppress nothing but a sha already reported. Dropping
+        them costs a full history re-read as soon as a branch reappears.
+        """
+        return current if current or not previous else previous
+
+    def complete_read(
+        self, current: Heads, unresolved: Collection[str], *, empty_confirmed: bool
+    ) -> bool:
+        """Whether this pass actually saw everything the repository offers.
+
+        An empty listing is never taken at face value the first time: trusting
+        it advances the cursor with no heads, and the idle gate then skips the
+        repository until somebody pushes to it. It counts only once the same
+        answer has been recorded before — which state written before this rule
+        existed never has. A head deliberately left out of the range (out of the
+        start window) is still a complete read.
+        """
+        if unresolved:
+            return False
+        return bool(current) or empty_confirmed
+
+    def empty_listing_confirmed(self, prior: Mapping[str, Any], field: str) -> bool:
+        return field in prior and not prior[field]
+
+    def cursor_value(self, prior: Mapping[str, Any], repo_updated_on: str, complete: bool) -> str:
+        return repo_updated_on if complete else str(prior.get("repo_updated_on") or "")
+
     def new_commits(
         self,
         repo: RepositoryRef,
         current_heads: Sequence[str],
         previous_heads: Sequence[str],
+        unresolved: set[str] | None = None,
     ) -> Iterable[Mapping[str, object]]:
-        try:
-            yield from self._client.commits_between(repo, current_heads, previous_heads)
-        except BitbucketApiError as exc:
-            if exc.status_code != 404 or not previous_heads:
-                raise
-            yield from self._client.commits_between(repo, current_heads, [])
+        includes = list(current_heads)
+        excludes = list(previous_heads)
+        # Every round either drops at least one sha or clears the excludes once,
+        # so this bounds the walk without ever stopping a walk that is still
+        # getting somewhere: giving up mid-repair leaves the repository to fail
+        # the same way on every future sync.
+        rounds = len(includes) + len(excludes) + 2
+        for attempt in range(rounds):
+            if attempt == RANGE_REPAIR_ATTEMPTS:
+                logger.warning(
+                    f"{repo.workspace}/{repo.slug}: commit range still being repaired after "
+                    f"{attempt} attempts; the branch listing is advertising heads the commits "
+                    "endpoint cannot resolve"
+                )
+            try:
+                yield from self._client.commits_between(repo, includes, excludes)
+                return
+            except BitbucketApiError as exc:
+                if exc.status_code != 404:
+                    raise
+                # Retrying re-yields whatever the failed attempt already
+                # emitted; bronze collapses the overlap on unique_key.
+                missing = exc.missing_shas
+                if not missing.intersection(includes) and not missing.intersection(excludes):
+                    if not excludes:
+                        raise
+                    excludes = []
+                    continue
+                if unresolved is not None:
+                    unresolved.update(missing.intersection(includes))
+                includes = [sha for sha in includes if sha not in missing]
+                excludes = [sha for sha in excludes if sha not in missing]
+                if not includes:
+                    return
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/metric_events.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/metric_events.py
index 8fc633eff..9f3f23de5 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/metric_events.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/metric_events.py
@@ -28,7 +28,7 @@ def repository_records(self, repo, bucket_id: int):
         )
         for record in records:
             identity = record.get("uuid") or record.get("id") or record.get("name")
-            if identity is None:
+            if identity is None or not self.include(record):
                 continue
             entity_key = unique_key(self._tenant_id, self._source_id, *repo_scope(repo), identity)
             entity_keys.add(entity_key)
@@ -59,6 +59,9 @@ def repository_records(self, repo, bucket_id: int):
     def enabled(self, repo) -> bool:
         return True
 
+    def include(self, record: Mapping[str, Any]) -> bool:
+        return True
+
     def project(self, record: Mapping[str, Any]) -> Mapping[str, Any]:
         return {}
 
@@ -83,6 +86,15 @@ class TagsStream(RepositorySnapshotStream):
     name = "tags"
     resource = "refs/tags"
 
+    def include(self, record: Mapping[str, Any]) -> bool:
+        # The tag's own date, not `target`'s: target is the commit it points at,
+        # and a tag cut today can reference a years-old commit. A tag carrying
+        # no date of its own is kept rather than judged by its commit.
+        tagged_at = str(record.get("date") or "")
+        if not self._start_date or not tagged_at:
+            return True
+        return tagged_at[:10] >= self._start_date
+
     def project(self, record: Mapping[str, Any]) -> Mapping[str, Any]:
         return {**record, "target_hash": (record.get("target") or {}).get("hash")}
 
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_comments.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_comments.py
index ab812c3af..877e4ef57 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_comments.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_comments.py
@@ -6,6 +6,24 @@
 from source_bitbucket_cloud.streams.base import repo_scope, schema, truncate, unique_key
 from source_bitbucket_cloud.streams.pr_base import PullRequestStateStream
 
+PR_COMMENT_FIELDS = ",".join(
+    [
+        "values.id",
+        "values.content.raw",
+        "values.created_on",
+        "values.updated_on",
+        "values.user.display_name",
+        "values.user.uuid",
+        "values.user.account_id",
+        "values.inline.path",
+        "values.inline.from",
+        "values.inline.to",
+        "values.parent.id",
+        "values.deleted",
+        "next",
+    ]
+)
+
 
 class PRCommentsStream(PullRequestStateStream):
     name = "pull_request_comments"
@@ -17,7 +35,9 @@ def pull_request_records(self, repo, pr: Mapping[str, Any]) -> Iterable[Mapping[
         generation = self.generation(repo.uuid, pr_id, "comments")
         entity_keys: set[str] = set()
         path = self._client.repo_path(repo, f"pullrequests/{pr_id}/comments")
-        present, comments = self._client.paginate_optional(path, params={"pagelen": "100"})
+        present, comments = self._client.paginate_optional(
+            path, params={"pagelen": "100", "fields": PR_COMMENT_FIELDS}
+        )
         for comment in comments:
             comment_id = comment.get("id")
             if comment_id is None:
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_commits.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_commits.py
index 67cf9adea..3ba3f478a 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_commits.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_commits.py
@@ -7,6 +7,9 @@
 from source_bitbucket_cloud.streams.pr_base import PullRequestStateStream
 
 
+PR_COMMIT_FIELDS = "values.hash,values.author.user.uuid,values.author.user.account_id,next"
+
+
 class PRCommitsStream(PullRequestStateStream):
     name = "pull_request_commits"
 
@@ -17,7 +20,9 @@ def pull_request_records(self, repo, pr: Mapping[str, Any]) -> Iterable[Mapping[
         generation = self.generation(repo.uuid, pr_id, "commits")
         entity_keys: set[str] = set()
         path = self._client.repo_path(repo, f"pullrequests/{pr_id}/commits")
-        present, commits = self._client.paginate_optional(path, params={"pagelen": "100"})
+        present, commits = self._client.paginate_optional(
+            path, params={"pagelen": "100", "fields": PR_COMMIT_FIELDS}
+        )
         for commit_order, commit in enumerate(commits):
             sha = str(commit.get("hash") or "")
             if not sha:
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_diffstat.py b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_diffstat.py
index 9384163e7..71203a250 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_diffstat.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/source_bitbucket_cloud/streams/pr_diffstat.py
@@ -3,6 +3,7 @@
 from collections.abc import Iterable, Mapping
 from typing import Any
 
+from source_bitbucket_cloud.client import UNCOMPUTABLE_DIFF
 from source_bitbucket_cloud.streams.base import repo_scope, schema, unique_key
 from source_bitbucket_cloud.streams.pr_base import PullRequestStateStream
 
@@ -17,7 +18,9 @@ def pull_request_records(self, repo, pr: Mapping[str, Any]) -> Iterable[Mapping[
         generation = self.generation(repo.uuid, pr_id, "diffstat")
         path = self._client.repo_path(repo, f"pullrequests/{pr_id}/diffstat")
         entity_keys: set[str] = set()
-        present, entries = self._client.paginate_optional(path, params={"pagelen": "100"})
+        present, entries = self._client.paginate_optional(
+            path, params={"pagelen": "100"}, tolerate_messages=UNCOMPUTABLE_DIFF
+        )
         for entry in entries:
             new_file = entry.get("new") or {}
             old_file = entry.get("old") or {}
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/conftest.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/conftest.py
index 47c58d0c3..273f5cf61 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/tests/conftest.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/conftest.py
@@ -31,8 +31,8 @@ def repository(slug: str = "repo", uuid: str = "{r-1}", **raw: Any) -> Repositor
 
 
 def branch(name: str = "main", sha: str = "a1", **raw: Any) -> BranchRef:
-    data = {"name": name, "target": {"hash": sha}, **raw}
-    return BranchRef(name, sha, "2026-06-01T00:00:00+00:00", name == "main", data)
+    target_date = str((raw.get("target") or {}).get("date") or "2026-06-01T00:00:00+00:00")
+    return BranchRef(name, sha, target_date, name == "main")
 
 
 class FakeCatalog:
@@ -60,6 +60,10 @@ def repositories(self) -> list[RepositoryRef]:
     def branches(self, repo: RepositoryRef) -> list[BranchRef]:
         return self._client.branches(repo) if self._client else []
 
+    @property
+    def branch_cache_size(self) -> tuple[int, int]:
+        return 0, 0
+
 
 class _FakeResponse:
     def __init__(self, body: Mapping[str, Any]):
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_base.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_base.py
index b0986987b..0d7aec4b4 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_base.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_base.py
@@ -6,6 +6,7 @@
 from source_bitbucket_cloud.client import BitbucketApiError, BitbucketClient
 from source_bitbucket_cloud.streams.base import (
     BUCKET_COUNT,
+    MAX_TEXT_BYTES,
     normalize_start_date,
     now_iso,
     repo_state_key,
@@ -22,7 +23,8 @@ def test_helpers():
     with pytest.raises(ValueError):
         normalize_start_date("invalid")
     assert truncate(None) is None
-    assert len(truncate("x" * 20_000).encode()) <= 16_384
+    assert len(truncate("x" * 20_000).encode()) <= MAX_TEXT_BYTES
+    assert MAX_TEXT_BYTES <= 2_048, "generated descriptions must not multiply bronze storage"
     assert unique_key("T", "S", "a:b") == "T:S:a%3Ab"
     assert 0 <= repository_bucket("{r-1}") < BUCKET_COUNT
 
@@ -97,3 +99,38 @@ def test_client_pagination_follows_next_and_detects_loops():
     pages["second"] = Response({"next": "first"}, url="second")
     with pytest.raises(RuntimeError, match="pagination loop"):
         list(client.paginate("first"))
+
+
+def test_state_survives_a_bucket_count_change(commits_stream):
+    """Keys are repository-scoped and the bucket is derived by hash at read
+    time, so state written under any bucket count must resume under any other —
+    discarding it would force the full resync the connector promises to avoid."""
+    stored = {
+        "version": 3,
+        "bucket_count": 4,
+        "repositories": {"ws/repo": {"head_shas": ["a"], "repo_updated_on": "d1"}},
+    }
+
+    commits_stream.state = stored
+
+    assert commits_stream.state["repositories"] == stored["repositories"]
+    assert commits_stream.state["bucket_count"] == BUCKET_COUNT
+
+
+def test_state_snapshot_is_isolated_from_later_commits(commits_stream):
+    """The platform serialises the state property while workers commit; the
+    snapshot it took must not change under its feet."""
+    first = repository(slug="one")
+    commits_stream.state = {}
+    commits_stream.commit_repository_state(first, {"head_shas": ["a"]})
+
+    snapshot = commits_stream.state
+    commits_stream.commit_repository_state(repository(slug="two", uuid="{r-2}"), {"head_shas": ["b"]})
+
+    assert list(snapshot["repositories"]) == [repo_state_key(first)]
+
+
+def test_incremental_streams_checkpoint_mid_bucket(commits_stream):
+    assert commits_stream.state_checkpoint_interval, (
+        "without an interval, state is only emitted per bucket and a crash re-reads hours of work"
+    )
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_branches.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_branches.py
index 23ce157d5..2df294682 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_branches.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_branches.py
@@ -2,7 +2,10 @@
 from conftest import SHARED, FakeCatalog, FakeClient, branch
 from source_bitbucket_cloud.client import BitbucketApiError
 from source_bitbucket_cloud.streams.base import repo_state_key, repository_bucket
-from source_bitbucket_cloud.streams.commit_branch_reachability import CommitBranchReachabilityStream
+from source_bitbucket_cloud.streams.commit_branch_reachability import (
+    RANGE_PREFETCH,
+    CommitBranchReachabilityStream,
+)
 
 
 def reachability_stream(repo, client):
@@ -50,12 +53,32 @@ def test_reachability_emits_commits_for_every_changed_branch(stream_args, client
 
 def test_reachability_records_deleted_branch(stream_args, client, repo):
     stream = CommitBranchReachabilityStream(**stream_args)
-    stream.state = {"version": 3, "bucket_count": 8, "repositories": {repo_state_key(repo): {"heads": {"release": "old"}}}}
-    client.branch_values[repo.uuid] = []
+    stream.state = {
+        "version": 3,
+        "bucket_count": 8,
+        "repositories": {repo_state_key(repo): {"heads": {"main": "m1", "release": "old"}}},
+    }
+    client.branch_values[repo.uuid] = [branch("main", "m1")]
     records = list(stream.read_records(SyncMode.incremental, stream_slice={"bucket_id": repository_bucket(repo_state_key(repo))}))
-    assert records[0]["branch_name"] == "release"
-    assert records[0]["reachability_action"] == "branch_deleted"
-    assert records[0]["commit_sha"] is None
+    deleted = [record for record in records if record["reachability_action"] == "branch_deleted"]
+    assert [record["branch_name"] for record in deleted] == ["release"]
+    assert deleted[0]["commit_sha"] is None
+
+
+def test_reachability_does_not_delete_every_branch_over_one_empty_listing(stream_args, client, repo):
+    """A listing that returns nothing would mark the whole repository deleted,
+    and a later listing that finds the branches again emits no correction."""
+    stream = CommitBranchReachabilityStream(**stream_args)
+    prior = {"heads": {"main": "m1", "release": "old"}, "repo_updated_on": "earlier"}
+    stream.state = {"version": 3, "bucket_count": 8, "repositories": {repo_state_key(repo): prior}}
+    client.branch_values[repo.uuid] = []
+
+    records = list(
+        stream.read_records(SyncMode.incremental, stream_slice={"bucket_id": repository_bucket(repo_state_key(repo))})
+    )
+
+    assert records == []
+    assert stream.state["repositories"][repo_state_key(repo)] == prior, "nothing read, nothing advanced"
 
 
 def test_reachability_moved_branch_emits_added_and_removed(repo):
@@ -117,3 +140,41 @@ def test_reachability_404_after_partial_page_does_not_re_emit(repo):
 
     assert ("added", "partial") not in emitted
     assert ("reset", "full") in emitted
+
+
+class _CountingRange(FakeClient):
+    """Counts how many commits the caller has pulled out of the range."""
+
+    def __init__(self, size):
+        super().__init__()
+        self.size = size
+        self.pulled = 0
+
+    def commits_between(self, repo, include, exclude):
+        self.commit_calls.append((list(include), list(exclude)))
+
+        def pages():
+            for index in range(self.size):
+                self.pulled += 1
+                yield {"hash": f"c{index}", "date": "2026-06-01"}
+
+        return pages()
+
+
+def test_reachability_does_not_hold_a_whole_history_in_memory(repo):
+    """A first read of a branch spans its entire history and several
+    repositories are read at once, so the range may not be materialised."""
+    history = RANGE_PREFETCH * 3
+    client = _CountingRange(history)
+    client.branch_values[repo.uuid] = [branch("main", "new")]
+    stream = reachability_stream(repo, client)
+    stream.state = {}
+
+    records = stream.read_records(
+        SyncMode.incremental, stream_slice={"bucket_id": repository_bucket(repo_state_key(repo))}
+    )
+    first = next(records)
+
+    assert first["commit_sha"] == "c0"
+    assert client.pulled <= RANGE_PREFETCH, f"buffered {client.pulled} of {history} before emitting anything"
+    assert len(list(records)) == history - 1, "and the rest of the range still arrives"
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_client.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_client.py
index b233dda82..f2edad865 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_client.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_client.py
@@ -4,7 +4,7 @@
 
 import pytest
 
-from source_bitbucket_cloud.client import BitbucketApiError, BitbucketClient
+from source_bitbucket_cloud.client import BitbucketApiError, BitbucketClient, BranchRef
 
 BASE = "https://api.bitbucket.org/2.0/"
 
@@ -114,7 +114,9 @@ def test_optional_absent_on_403(self):
         assert present is False
         assert list(records) == []
 
-    def test_optional_stops_gracefully_on_later_404(self):
+    def test_optional_refuses_to_truncate_a_started_collection(self):
+        """Ending the collection here would publish part of a snapshot as the
+        whole of it, deleting whatever the unread pages held."""
         client = make_client(
             [
                 Response(body={"values": [{"n": 1}], "next": BASE + "x?page=2"}),
@@ -123,7 +125,22 @@ def test_optional_stops_gracefully_on_later_404(self):
         )
         present, records = client.paginate_optional("repositories/ws/pipelines")
         assert present is True
-        assert [row["n"] for row in records] == [1]
+        with pytest.raises(RuntimeError, match="continuation page"):
+            list(records)
+
+    def test_a_refused_continuation_is_not_a_denial(self):
+        """A denial marks the repository inaccessible for every later stream;
+        one bad page must not."""
+        client = make_client(
+            [
+                Response(body={"values": [{"n": 1}], "next": BASE + "x?page=2"}),
+                Response(403),
+            ]
+        )
+        _, records = client.paginate_optional("repositories/ws/pipelines")
+        with pytest.raises(RuntimeError) as raised:
+            list(records)
+        assert not isinstance(raised.value, BitbucketApiError)
 
 
 class TestFieldMapping:
@@ -175,3 +192,29 @@ def test_branches_maps_and_marks_default(self):
         assert branches[0].is_default is True
         assert branches[0].head_sha == "a1"
         assert branches[1].is_default is False
+
+
+class TestBranchRefStaysSlim:
+    """The catalog holds every branch of every repository for a whole sync."""
+
+    def test_only_the_fields_the_streams_read_are_kept(self):
+        ref = BranchRef(name="main", head_sha="a1", target_date=None, is_default=True)
+
+        assert BranchRef.__slots__ == ("name", "head_sha", "target_date", "is_default")
+        assert not hasattr(ref, "__dict__"), "a per-instance dict would dwarf the four fields"
+
+
+def test_a_credential_failure_on_a_later_page_is_not_wrapped():
+    """401 has to reach the sync as itself: it aborts the whole read with the
+    cause instead of quarantining every remaining repository one at a time."""
+    client = make_client(
+        [
+            Response(body={"values": [{"n": 1}], "next": BASE + "x?page=2"}),
+            Response(401),
+        ]
+    )
+    _, records = client.paginate_optional("repositories/ws/pipelines")
+
+    with pytest.raises(BitbucketApiError) as raised:
+        list(records)
+    assert raised.value.status_code == 401
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_commits.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_commits.py
index 62997b585..c17da7dbc 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_commits.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_commits.py
@@ -1,6 +1,6 @@
 from airbyte_cdk.models import SyncMode
 from conftest import branch
-from source_bitbucket_cloud.streams.base import repo_state_key, repository_bucket
+from source_bitbucket_cloud.streams.base import MAX_TEXT_BYTES, repo_state_key, repository_bucket
 
 
 def commit(sha="c1", date="2026-06-01T00:00:00+00:00", **extra):
@@ -60,6 +60,6 @@ def test_start_date_filters_old_commits(commits_stream, client, repo):
 
 def test_message_truncation_and_raw_identity(commits_stream, repo):
     record = commits_stream._record(repo, commit(message="x" * 20_000, author={"raw": "buildbot", "user": None}))
-    assert len(record["message"].encode()) <= 16_384
+    assert len(record["message"].encode()) <= MAX_TEXT_BYTES
     assert record["author_name"] == "buildbot"
     assert record["author_email"] is None
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.py
new file mode 100644
index 000000000..4f0496397
--- /dev/null
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_concurrency.py
@@ -0,0 +1,371 @@
+from __future__ import annotations
+
+import threading
+import time
+from contextlib import closing
+
+import pytest
+from source_bitbucket_cloud.client import BitbucketApiError, RepositoryRef
+from source_bitbucket_cloud.streams.base import (
+    BUCKET_COUNT,
+    QUEUE_POLL_SECONDS,
+    RECORD_BUFFER,
+    repo_state_key,
+    repository_bucket,
+)
+from source_bitbucket_cloud.streams.commits import CommitsStream
+from tests.conftest import SHARED, FakeCatalog, FakeClient, branch, repository
+
+DATE = "2026-06-01T00:00:00+00:00"
+VOLATILE = {"collected_at", "generation_id", "unique_key"}
+
+
+def fleet(size: int = 12):
+    return [repository(slug=f"repo{index:02d}", uuid=f"{{r-{index}}}") for index in range(size)]
+
+
+def fleet_in_one_bucket(size: int, bucket: int = 0) -> list[RepositoryRef]:
+    """Repositories that all land in the same slice, so one read_records call
+    really does run several workers."""
+    repos: list[RepositoryRef] = []
+    for index in range(size * BUCKET_COUNT * 20):
+        repo = repository(slug=f"same{index:03d}", uuid=f"{{s-{index}}}")
+        if repository_bucket(repo_state_key(repo)) == bucket:
+            repos.append(repo)
+        if len(repos) == size:
+            return repos
+    raise AssertionError(f"could not place {size} repositories in bucket {bucket}")
+
+
+class FleetClient(FakeClient):
+    """One branch and one commit per repository, so a record identifies its
+    repository unambiguously."""
+
+    def __init__(self, repos, delay: float = 0.0):
+        super().__init__()
+        self.delay = delay
+        self.threads: set[int] = set()
+        self._lock = threading.Lock()
+        for repo in repos:
+            self.branch_values[repo.uuid] = [branch("main", f"head-{repo.slug}")]
+
+    def branches(self, repo):
+        with self._lock:
+            self.threads.add(threading.get_ident())
+        if self.delay:
+            time.sleep(self.delay)
+        return self.branch_values.get(repo.uuid, [])
+
+    def commits_between(self, repo, include, exclude):
+        with self._lock:
+            self.commit_calls.append((list(include), list(exclude)))
+        return iter([{"hash": sha, "date": DATE} for sha in include])
+
+
+def read_all_buckets(stream):
+    records = []
+    for bucket in range(BUCKET_COUNT):
+        records.extend(stream.read_records(None, stream_slice={"bucket_id": bucket}))
+    return records
+
+
+def build(repos, client, concurrency: int):
+    stream = CommitsStream(
+        **{**SHARED, "concurrency": concurrency, "client": client, "catalog": FakeCatalog(repos, client)}
+    )
+    stream.state = {}
+    return stream
+
+
+def comparable(records):
+    return sorted(
+        tuple(sorted((k, str(v)) for k, v in record.items() if k not in VOLATILE)) for record in records
+    )
+
+
+class TestConcurrentReadsMatchSerialOnes:
+    @pytest.mark.parametrize("concurrency", [2, 4, 8])
+    def test_same_records_and_same_state(self, concurrency):
+        repos = fleet()
+        serial_client = FleetClient(repos)
+        serial = build(repos, serial_client, 1)
+        expected = read_all_buckets(serial)
+
+        parallel_client = FleetClient(repos)
+        parallel = build(repos, parallel_client, concurrency)
+        actual = read_all_buckets(parallel)
+
+        assert comparable(actual) == comparable(expected)
+        assert parallel.state == serial.state
+        assert len(parallel_client.commit_calls) == len(serial_client.commit_calls)
+
+    def test_a_slow_repository_does_not_hold_back_finished_ones(self):
+        """One deep history at the front of a bucket must not park the whole
+        pool: the other repositories' records leave as soon as they are ready."""
+        repos = fleet_in_one_bucket(4)
+        slow = repos[0]
+        gate = threading.Event()
+
+        class SlowFirstClient(FleetClient):
+            def branches(self, repo):
+                if repo.uuid == slow.uuid:
+                    gate.wait(timeout=30)
+                return super().branches(repo)
+
+        client = SlowFirstClient(repos)
+        stream = build(repos, client, 4)
+        records = stream.read_records(None, stream_slice={"bucket_id": 0})
+
+        fast_slugs = {repo.slug for repo in repos[1:]}
+        seen: set[str] = set()
+        for record in records:
+            seen.add(record["repo_slug"])
+            if seen >= fast_slugs:
+                break
+        assert seen >= fast_slugs and slow.slug not in seen, (
+            "finished repositories must drain while the slow one is still fetching"
+        )
+
+        gate.set()
+        seen.update(record["repo_slug"] for record in records)
+        assert slow.slug in seen, "and the slow repository still completes"
+
+    def test_work_actually_runs_in_parallel(self):
+        repos = fleet()
+        client = FleetClient(repos, delay=0.002)
+
+        read_all_buckets(build(repos, client, 8))
+
+        assert len(client.threads) > 1, "the pool must do the fetching, not the consumer"
+
+    def test_one_worker_stays_on_the_serial_path(self):
+        repos = fleet()
+        client = FleetClient(repos)
+
+        read_all_buckets(build(repos, client, 1))
+
+        assert client.threads == {threading.get_ident()}
+
+
+class TestFailuresKeepTheirSemantics:
+    def denied_client(self, repos, victim: str, status: int):
+        class DeniedClient(FleetClient):
+            def branches(self, repo):
+                if repo.slug == victim:
+                    raise BitbucketApiError(status, "https://api.bitbucket.org/2.0/x", "denied")
+                return super().branches(repo)
+
+        return DeniedClient(repos)
+
+    @pytest.mark.parametrize("status", [403, 404])
+    def test_a_denied_repository_is_skipped_not_failed(self, status):
+        repos = fleet()
+        client = self.denied_client(repos, "repo05", status)
+        stream = build(repos, client, 4)
+
+        records = read_all_buckets(stream)
+
+        assert stream._failed_repositories == [], f"HTTP {status} must not count as a failure"
+        assert "ws/repo05" in stream._skipped_repositories, f"HTTP {status} must be recorded as skipped"
+        assert {r["repo_slug"] for r in records} == {repo.slug for repo in repos} - {"repo05"}, (
+            f"every other repository must still be read past an HTTP {status}"
+        )
+
+    def test_a_transient_failure_still_fails_the_sync(self):
+        repos = fleet()
+        client = self.denied_client(repos, "repo05", 500)
+        stream = build(repos, client, 4)
+
+        with pytest.raises(RuntimeError, match="repositories failed"):
+            read_all_buckets(stream)
+
+        assert stream._failed_repositories == ["ws/repo05"]
+        healthy = [repo for repo in repos if repo.slug != "repo05"]
+        assert all(repo_state_key(repo) in stream.state["repositories"] for repo in healthy), (
+            "one repository's failure must not cost its neighbours their checkpoints"
+        )
+
+    def test_a_credential_failure_aborts(self):
+        repos = fleet()
+        client = self.denied_client(repos, "repo05", 401)
+        stream = build(repos, client, 4)
+
+        with pytest.raises(RuntimeError, match="authentication failed"):
+            read_all_buckets(stream)
+
+    def test_a_failed_repository_does_not_advance_its_state(self):
+        repos = fleet()
+        client = self.denied_client(repos, "repo05", 500)
+        stream = build(repos, client, 4)
+
+        with pytest.raises(RuntimeError):
+            read_all_buckets(stream)
+
+        victim = next(repo for repo in repos if repo.slug == "repo05")
+        assert repo_state_key(victim) not in stream.state["repositories"]
+
+
+class TestBackpressure:
+    def test_a_worker_stops_at_the_buffer_instead_of_reading_ahead(self):
+        """Two workers in one slice, each with more history than the buffer
+        holds: they must park rather than grow, and lose nothing by parking."""
+        repos = fleet_in_one_bucket(2)
+        overflow = RECORD_BUFFER * 3
+
+        class WideClient(FleetClient):
+            def __init__(self, fleet_repos: list[RepositoryRef]) -> None:
+                super().__init__(fleet_repos)
+                self.produced = 0
+
+            def commits_between(self, repo, include, exclude):
+                def history():
+                    for index in range(overflow):
+                        with self._lock:
+                            self.produced += 1
+                        yield {"hash": f"{repo.slug}-{index}", "date": DATE}
+
+                return history()
+
+        client = WideClient(repos)
+        stream = build(repos, client, 2)
+        records = stream.read_records(None, stream_slice={"bucket_id": 0})
+
+        next(records)
+        parked = client.produced
+        assert parked <= 2 * (RECORD_BUFFER + 1), (
+            f"{parked} records fetched before the first was consumed; two buffers hold "
+            f"{2 * RECORD_BUFFER}"
+        )
+
+        assert len(list(records)) + 1 == overflow * len(repos), "parking must not drop records"
+
+
+class TestAbandoningTheReadTerminates:
+    def test_closing_the_generator_releases_parked_workers(self):
+        """Airbyte can stop reading mid-bucket; workers blocked on a full
+        buffer must be released before the pool is joined."""
+        repos = fleet_in_one_bucket(6)
+        overflow = RECORD_BUFFER * 2
+
+        class WideClient(FleetClient):
+            def commits_between(self, repo, include, exclude):
+                return iter([{"hash": f"{repo.slug}-{n}", "date": DATE} for n in range(overflow)])
+
+        stream = build(repos, WideClient(repos), 4)
+        finished = threading.Event()
+
+        def read_a_little():
+            records = stream.read_records(None, stream_slice={"bucket_id": 0})
+            for _ in range(3):
+                next(records, None)
+            records.close()
+            finished.set()
+
+        reader = threading.Thread(target=read_a_little, daemon=True)
+        reader.start()
+        reader.join(timeout=30)
+
+        assert finished.is_set(), "closing the read must not hang on parked workers"
+
+
+class TestStateFollowsTheRecords:
+    """A checkpoint can be taken between any two records the consumer emits, so
+    state that claims a repository before its records have left would lose them
+    to a crash in that window — and the idle gate would skip it next sync."""
+
+    def test_state_is_not_published_while_records_are_still_queued(self):
+        repos = fleet_in_one_bucket(1)
+        repo = repos[0]
+
+        class ThreeCommits(FleetClient):
+            def commits_between(self, repo, include, exclude):
+                return iter([{"hash": f"c{index}", "date": DATE} for index in range(3)])
+
+        stream = build(repos, ThreeCommits(repos), 4)
+        records = stream.read_records(None, stream_slice={"bucket_id": 0})
+
+        next(records)
+        assert stream.state["repositories"] == {}, (
+            "the worker finished fetching, but its records have not been emitted yet"
+        )
+
+        assert len(list(records)) == 2
+        assert repo_state_key(repo) in stream.state["repositories"], "and state lands once they have"
+
+    def test_every_repository_still_checkpoints_by_the_end(self):
+        repos = fleet()
+        client = FleetClient(repos)
+        stream = build(repos, client, 4)
+
+        read_all_buckets(stream)
+
+        assert set(stream.state["repositories"]) == {repo_state_key(repo) for repo in repos}
+
+    def test_an_abandoned_read_publishes_no_state(self):
+        repos = fleet_in_one_bucket(2)
+        overflow = RECORD_BUFFER * 2
+
+        class WideClient(FleetClient):
+            def commits_between(self, repo, include, exclude):
+                return iter([{"hash": f"{repo.slug}-{index}", "date": DATE} for index in range(overflow)])
+
+        stream = build(repos, WideClient(repos), 2)
+        records = stream.read_records(None, stream_slice={"bucket_id": 0})
+        next(records)
+        records.close()
+
+        assert stream.state["repositories"] == {}, "nothing drained, nothing claimed"
+
+
+class TestOutputRotatesBetweenRepositories:
+    def test_one_fast_producer_does_not_hold_the_floor(self):
+        """A producer that refills between yields would otherwise emit its whole
+        repository first, leaving every other worker parked on a full buffer."""
+        repos = fleet_in_one_bucket(2)
+        history = 1_500
+
+        class FastClient(FleetClient):
+            def commits_between(self, repo, include, exclude):
+                return iter([{"hash": f"{repo.slug}-{index}", "date": DATE} for index in range(history)])
+
+        stream = build(repos, FastClient(repos), 2)
+        opening: list[str] = []
+
+        # A fast consumer empties each queue before the producer refills, testing nothing.
+        # Closed explicitly: this test fails while holding the generator, which parks workers.
+        with closing(stream.read_records(None, stream_slice={"bucket_id": 0})) as records:
+            for record in records:
+                opening.append(record["repo_slug"])
+                time.sleep(0.0001)
+                if len(opening) == history // 2:
+                    break
+
+        assert len(set(opening)) == 2, (
+            f"the first {len(opening)} records all came from {opening[0]}; output must rotate"
+        )
+
+
+class TestASlowConsumerIsNotAnAbsentOne:
+    def test_a_paused_consumer_still_receives_every_record(self):
+        """A full queue means the destination is slow, not that the read was
+        abandoned: the worker parks and delivers once consumption resumes."""
+        repos = fleet_in_one_bucket(1)
+        history = RECORD_BUFFER * 3
+
+        class WideClient(FleetClient):
+            def commits_between(self, repo, include, exclude):
+                return iter([{"hash": f"c{index}", "date": DATE} for index in range(history)])
+
+        stream = build(repos, WideClient(repos), 2)
+        seen = 0
+
+        with closing(stream.read_records(None, stream_slice={"bucket_id": 0})) as records:
+            for record in records:
+                assert record["record_type"] == "item"
+                if seen == 0:
+                    # Long enough for the worker to fill the buffer and park.
+                    time.sleep(QUEUE_POLL_SECONDS * 3)
+                seen += 1
+
+        assert seen == history, f"parked worker delivered {seen} of {history}"
+        assert repo_state_key(repos[0]) in stream.state["repositories"]
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_head_retention.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_head_retention.py
new file mode 100644
index 000000000..8414c412d
--- /dev/null
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_head_retention.py
@@ -0,0 +1,234 @@
+from __future__ import annotations
+
+import pytest
+
+from source_bitbucket_cloud.streams.base import BUCKET_COUNT, STATE_VERSION, repo_state_key
+from source_bitbucket_cloud.streams.branches import BranchesStream
+from source_bitbucket_cloud.streams.commit_branch_reachability import CommitBranchReachabilityStream
+from source_bitbucket_cloud.streams.commits import CommitsStream
+from source_bitbucket_cloud.streams.file_changes import FileChangesStream
+from tests.conftest import SHARED, FakeCatalog, FakeClient, branch, repository
+
+DATE = "2026-06-01T00:00:00+00:00"
+HEAD_FIELD = {
+    CommitsStream: "head_shas",
+    FileChangesStream: "head_shas",
+    CommitBranchReachabilityStream: "heads",
+}
+
+
+def read_all_buckets(stream):
+    records = []
+    for bucket in range(BUCKET_COUNT):
+        records.extend(stream.read_records(None, stream_slice={"bucket_id": bucket}))
+    return records
+
+
+def synced_state(repo, field, value, updated_on):
+    return {
+        "version": STATE_VERSION,
+        "bucket_count": BUCKET_COUNT,
+        "repositories": {repo_state_key(repo): {field: value, "repo_updated_on": updated_on}},
+    }
+
+
+@pytest.mark.parametrize("stream_class", list(HEAD_FIELD))
+class TestEmptyListingDoesNotForgetHeads:
+    def build(self, stream_class, client, repo):
+        return stream_class(**{**SHARED, "client": client, "catalog": FakeCatalog([repo], client)})
+
+    def known(self, stream_class):
+        return ["known"] if HEAD_FIELD[stream_class] == "head_shas" else {"main": "known"}
+
+    def test_heads_survive_a_listing_that_returns_nothing(self, stream_class, repo):
+        field = HEAD_FIELD[stream_class]
+        client = FakeClient()
+        client.branch_values[repo.uuid] = []
+        stream = self.build(stream_class, client, repo)
+        stream.state = synced_state(repo, field, self.known(stream_class), "older")
+
+        read_all_buckets(stream)
+
+        stored = stream.state["repositories"][repo_state_key(repo)]
+        assert stored[field] == self.known(stream_class), (
+            "an empty listing must not cost the exclude set — the next range would re-read all history"
+        )
+        assert stored["repo_updated_on"] == "older", (
+            "advancing the cursor over an empty listing would gate away whatever the push carried"
+        )
+
+    def test_the_listing_is_retried_even_if_nothing_is_pushed_after_it(self, stream_class, repo):
+        """The empty answer may have been the API's, not the repository's. The
+        next pass must look again rather than trust an idle gate that was
+        closed by a read which saw nothing."""
+        field = HEAD_FIELD[stream_class]
+        client = FakeClient()
+        client.branch_values[repo.uuid] = []
+        stream = self.build(stream_class, client, repo)
+        stream.state = synced_state(repo, field, self.known(stream_class), "older")
+        read_all_buckets(stream)
+
+        client.branch_values[repo.uuid] = [branch("main", "fresh")]
+        client.commit_values = [{"hash": "fresh", "date": DATE}]
+        retried = self.build(stream_class, client, repo)
+        retried.state = stream.state
+        client.commit_calls.clear()
+
+        read_all_buckets(retried)
+
+        assert client.commit_calls, "the repository must be looked at again"
+        assert all(excludes for _, excludes in client.commit_calls), "and diffed against the retained head"
+
+    def test_a_reappearing_branch_is_diffed_not_re_read(self, stream_class, repo):
+        field = HEAD_FIELD[stream_class]
+        client = FakeClient()
+        client.branch_values[repo.uuid] = []
+        stream = self.build(stream_class, client, repo)
+        stream.state = synced_state(repo, field, self.known(stream_class), "older")
+        read_all_buckets(stream)
+
+        pushed = repository(updated_on="2026-07-01T00:00:00+00:00")
+        client.branch_values[pushed.uuid] = [branch("main", "fresh")]
+        client.commit_values = [{"hash": "fresh", "date": DATE}]
+        revived = self.build(stream_class, client, pushed)
+        revived.state = stream.state
+        client.commit_calls.clear()
+        read_all_buckets(revived)
+
+        assert client.commit_calls, "the revived branch must be fetched"
+        assert all(excludes for _, excludes in client.commit_calls), (
+            "every range must carry the retained head as an exclude"
+        )
+
+    def test_a_populated_listing_still_replaces_the_stored_heads(self, stream_class, repo):
+        field = HEAD_FIELD[stream_class]
+        client = FakeClient()
+        client.branch_values[repo.uuid] = [branch("main", "moved")]
+        client.commit_values = [{"hash": "moved", "date": DATE}]
+        stream = self.build(stream_class, client, repo)
+        stream.state = synced_state(repo, field, self.known(stream_class), "older")
+
+        read_all_buckets(stream)
+
+        stored = stream.state["repositories"][repo_state_key(repo)][field]
+        assert stored == (["moved"] if field == "head_shas" else {"main": "moved"})
+
+
+class TestBranchSnapshotsSurviveAnEmptyListing:
+    """A branch snapshot replaces the previous one, so publishing an empty one
+    deletes every branch the repository had."""
+
+    def build(self, client, repo):
+        return BranchesStream(**{**SHARED, "client": client, "catalog": FakeCatalog([repo], client)})
+
+    def synced(self, client, repo):
+        stream = self.build(client, repo)
+        stream.state = {}
+        read_all_buckets(stream)
+        return stream.state
+
+    def test_an_empty_listing_publishes_no_snapshot(self, repo):
+        client = FakeClient()
+        client.branch_values[repo.uuid] = [branch("main", "a1")]
+        state = self.synced(client, repo)
+
+        client.branch_values[repo.uuid] = []
+        pushed = repository(updated_on="2026-07-01T00:00:00+00:00")
+        second = self.build(client, pushed)
+        second.state = state
+
+        records = read_all_buckets(second)
+
+        assert records == [], "neither items nor a marker: the snapshot would read as 'no branches'"
+        assert second.state["repositories"][repo_state_key(pushed)]["repo_updated_on"] != "2026-07-01T00:00:00+00:00", (
+            "and the cursor must stay open so the listing is retried"
+        )
+
+    def test_an_empty_repository_publishes_once_the_answer_repeats(self, repo):
+        """A second consecutive empty listing is the repository, not the API."""
+        client = FakeClient()
+        client.branch_values[repo.uuid] = []
+        first = self.build(client, repo)
+        first.state = {}
+
+        assert read_all_buckets(first) == [], "one empty answer is only an observation"
+
+        second = self.build(client, repo)
+        second.state = first.state
+        records = read_all_buckets(second)
+
+        markers = [r for r in records if r["record_type"] == "snapshot_complete"]
+        assert markers and markers[0]["snapshot_item_count"] == 0
+        assert markers[0]["snapshot_available"] is True
+
+    def test_state_written_before_this_rule_is_not_trusted_into_a_deletion(self, repo):
+        """Deployed state carries no branch count, so a first empty listing
+        under the new code must not read as 'this repository has no branches'."""
+        client = FakeClient()
+        client.branch_values[repo.uuid] = []
+        stream = self.build(client, repo)
+        stream.state = {
+            "version": STATE_VERSION,
+            "bucket_count": BUCKET_COUNT,
+            "repositories": {repo_state_key(repo): {"repo_updated_on": "older"}},
+        }
+
+        records = read_all_buckets(stream)
+
+        assert records == []
+        assert stream.state["repositories"][repo_state_key(repo)]["repo_updated_on"] == "older", (
+            "the cursor must stay open so the listing is retried"
+        )
+
+
+@pytest.mark.parametrize("stream_class", list(HEAD_FIELD))
+class TestAFirstEmptyListingIsNotTrusted:
+    """On fresh state an empty listing is indistinguishable from a glitch, and
+    trusting it advances the cursor with no heads — the idle gate then skips the
+    repository until somebody pushes to it."""
+
+    def build(self, stream_class, client, repo):
+        stream = stream_class(**{**SHARED, "client": client, "catalog": FakeCatalog([repo], client)})
+        stream.state = {}
+        return stream
+
+    def test_the_cursor_stays_open(self, stream_class, repo):
+        client = FakeClient()
+        client.branch_values[repo.uuid] = []
+
+        stream = self.build(stream_class, client, repo)
+        read_all_buckets(stream)
+
+        stored = stream.state["repositories"][repo_state_key(repo)]
+        assert stored["repo_updated_on"] == "", "an unconfirmed empty listing must not close the idle gate"
+
+    def test_the_repository_is_read_again_without_a_push(self, stream_class, repo):
+        client = FakeClient()
+        client.branch_values[repo.uuid] = []
+        first = self.build(stream_class, client, repo)
+        read_all_buckets(first)
+
+        client.branch_values[repo.uuid] = [branch("main", "fresh")]
+        client.commit_values = [{"hash": "fresh", "date": DATE}]
+        second = self.build(stream_class, client, repo)
+        second.state = first.state
+
+        records = read_all_buckets(second)
+
+        assert records, "the reappearing branch must be picked up"
+        assert client.commit_calls, "and its range actually fetched"
+
+    def test_a_repeated_empty_listing_finally_settles(self, stream_class, repo):
+        client = FakeClient()
+        client.branch_values[repo.uuid] = []
+        first = self.build(stream_class, client, repo)
+        read_all_buckets(first)
+
+        second = self.build(stream_class, client, repo)
+        second.state = first.state
+        read_all_buckets(second)
+
+        stored = second.state["repositories"][repo_state_key(repo)]
+        assert stored["repo_updated_on"] == repo.raw["updated_on"], (
+            "the same answer twice is the repository, not the API"
+        )
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_inaccessible_repos.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_inaccessible_repos.py
index f27530103..aeee57eb6 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_inaccessible_repos.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_inaccessible_repos.py
@@ -1,11 +1,10 @@
 """A repository the token cannot read must not fail the sync.
 
 A repository can be listed for a workspace and still answer 403 to every request
-under it — routine with repo-scoped tokens and per-repository permissions, and
-observed in production. Retrying never changes it, so treating it as a failure
-leaves the sync red on every run and buries the transient failures that do need
-attention. These tests pin the distinction: denied is skipped, everything else
-still fails loudly.
+under it — routine with repo-scoped tokens and per-repository permissions.
+Retrying never changes it, so treating it as a failure leaves the sync red on
+every run and buries the transient failures that do need attention. These tests
+pin the distinction: denied is skipped, everything else still fails loudly.
 
 The matrix classes at the bottom run EVERY stream the source wires — the list is
 derived from `SourceBitbucketCloud.streams()` itself, so a stream added later is
@@ -26,7 +25,7 @@
 
 
 def every_stream_class():
-    """All stream classes, from the production wiring — not a hand list."""
+    """All stream classes, from the source's own wiring — not a hand list."""
     source = SourceBitbucketCloud()
     streams = source.streams(
         {
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_pr_children.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_pr_children.py
index 5fc5a00f3..4d29df30b 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_pr_children.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_pr_children.py
@@ -1,5 +1,7 @@
 from source_bitbucket_cloud.streams.metric_events import IssuesStream, PipelinesStream
 from source_bitbucket_cloud.streams.pr_activity import PRActivityStream
+from source_bitbucket_cloud.streams.pr_comments import PR_COMMENT_FIELDS
+from source_bitbucket_cloud.streams.pr_commits import PR_COMMIT_FIELDS
 from source_bitbucket_cloud.streams.pr_diffstat import PRDiffstatStream
 
 
@@ -140,3 +142,39 @@ def test_empty_pipeline_and_issue_results_keep_provider_watermark(stream_args, c
     issue_state = {"updated_on": "2026-06-02T00:00:00+00:00"}
     assert pipelines.pipeline_candidates(repo, pipeline_state)[2]["created_on"] == pipeline_state["created_on"]
     assert issues.selected_issues(repo, issue_state)[2] == issue_state
+
+
+class TestChildProjectionsCoverWhatTheStreamsRead:
+    """The API silently drops a misspelled fields entry, so a wrong projection
+    surfaces as NULL columns, not an error."""
+
+    def test_pr_commit_fields(self):
+        projected = set(PR_COMMIT_FIELDS.split(","))
+
+        assert "next" in projected, "without it pagination stops after one page"
+        assert projected == {
+            "values.hash",
+            "values.author.user.uuid",
+            "values.author.user.account_id",
+            "next",
+        }
+
+    def test_pr_comment_fields(self):
+        projected = set(PR_COMMENT_FIELDS.split(","))
+
+        assert "next" in projected, "without it pagination stops after one page"
+        assert projected == {
+            "values.id",
+            "values.content.raw",
+            "values.created_on",
+            "values.updated_on",
+            "values.user.display_name",
+            "values.user.uuid",
+            "values.user.account_id",
+            "values.inline.path",
+            "values.inline.from",
+            "values.inline.to",
+            "values.parent.id",
+            "values.deleted",
+            "next",
+        }
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_pull_requests.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_pull_requests.py
index cf9b7ebfa..8cfe84709 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_pull_requests.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_pull_requests.py
@@ -1,5 +1,5 @@
 from airbyte_cdk.models import SyncMode
-from source_bitbucket_cloud.streams.base import repo_state_key, repository_bucket
+from source_bitbucket_cloud.streams.base import MAX_TEXT_BYTES, repo_state_key, repository_bucket
 
 
 def pr(pr_id=42, updated_on="2026-06-30T01:00:00+00:00", **extra):
@@ -97,4 +97,4 @@ def spy(path, **kwargs):
 
 def test_description_is_bounded(pull_requests_stream, repo):
     record = pull_requests_stream._record(repo, pr(description="x" * 20_000))
-    assert len(record["description"].encode()) <= 16_384
+    assert len(record["description"].encode()) <= MAX_TEXT_BYTES
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_reliability.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_reliability.py
index bb71f4baa..ed4317c2f 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_reliability.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_reliability.py
@@ -151,9 +151,29 @@ def fake_paginate(path, *, params=None, method="GET", data=None, **kwargs):
         client.paginate = fake_paginate
         list(client.commits_between(repository(), ["new1", "new2"], ["old1"]))
         assert seen["method"] == "POST"
-        assert seen["params"] == {"pagelen": "100"}
+        assert seen["params"]["pagelen"] == "100"
         assert ("include", "new1") in seen["data"] and ("exclude", "old1") in seen["data"]
 
+    def test_commits_between_projects_only_what_the_streams_read(self):
+        """The API silently drops a misspelled fields entry, so a wrong list
+        surfaces as NULL columns, not an error: pin what the record builder and
+        the pagination need."""
+        projected = set(BitbucketClient.COMMIT_FIELDS.split(","))
+
+        assert "next" in projected, "without it pagination stops after one page"
+        record_reads = {
+            "values.hash",
+            "values.date",
+            "values.message",
+            "values.parents.hash",
+        }
+        identity_reads = {
+            f"values.{role}.{part}"
+            for role in ("author", "committer")
+            for part in ("raw", "user.display_name", "user.uuid", "user.account_id")
+        }
+        assert projected == record_reads | identity_reads | {"next"}
+
     def test_commits_between_without_current_heads_asks_nothing(self):
         client = self.make_client()
         calls = []
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_source.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_source.py
index 98e789961..a5039dbf8 100644
--- a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_source.py
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_source.py
@@ -49,35 +49,48 @@ def test_check_connection_reports_transport_errors(client_type):
     assert reason == "Bitbucket API request failed: offline"
 
 
+TRANSFORMED_STREAMS = [
+    "repositories",
+    "branches",
+    "pull_requests",
+    "pull_request_commits",
+    "pull_request_comments",
+    "pull_request_activity",
+    "pull_request_diffstat",
+    "commits",
+    "file_changes",
+    "commit_branch_reachability",
+]
+# Their first read of a repository pages whatever history it holds; everything
+# before them is bounded by a watermark.
+UNBOUNDED_STREAMS = ["commits", "file_changes", "commit_branch_reachability"]
+
+
 def test_streams_are_independent_and_share_client_and_catalog():
     streams = SourceBitbucketCloud().streams(CONFIG)
-    assert [stream.name for stream in streams] == [
-        "repositories",
-        "branches",
-        "pull_requests",
-        "pull_request_diffstat",
-        "pull_request_activity",
-        "pull_request_tasks",
-        "pull_request_comments",
-        "pull_request_commits",
-        "pipelines",
-        "pipeline_steps",
-        "pipeline_step_test_reports",
-        "deployments",
-        "environments",
-        "tags",
-        "issues",
-        "issue_comments",
-        "issue_changes",
-        "commits",
-        "commit_branch_reachability",
-        "file_changes",
-    ]
+    assert [stream.name for stream in streams] == TRANSFORMED_STREAMS
     assert len({id(stream._client) for stream in streams}) == 1
     assert len({id(stream._catalog) for stream in streams}) == 1
     assert not any(hasattr(stream, "parent") for stream in streams)
 
 
+def test_streams_the_transform_layer_reads_come_first():
+    """A sync that runs out of time must still have produced what dbt builds
+    on; the trailing streams have no model reading them."""
+    names = [stream.name for stream in SourceBitbucketCloud().streams(CONFIG)]
+
+    assert names[: len(TRANSFORMED_STREAMS)] == TRANSFORMED_STREAMS
+
+
+def test_watermark_bounded_streams_run_before_history_sized_ones():
+    """A stream that can page a whole history must not be able to starve the
+    streams that cannot."""
+    names = [stream.name for stream in SourceBitbucketCloud().streams(CONFIG)]
+    bounded = [name for name in TRANSFORMED_STREAMS if name not in UNBOUNDED_STREAMS]
+
+    assert max(names.index(name) for name in bounded) < min(names.index(name) for name in UNBOUNDED_STREAMS)
+
+
 def test_tenant_identity_and_spec():
     streams = SourceBitbucketCloud().streams(CONFIG)
     assert all(stream._tenant_id == "T" and stream._source_id == "S" for stream in streams)
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_start_date_window.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_start_date_window.py
new file mode 100644
index 000000000..cbb7470a4
--- /dev/null
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_start_date_window.py
@@ -0,0 +1,230 @@
+from __future__ import annotations
+
+import pytest
+
+from source_bitbucket_cloud.client import BranchRef
+from source_bitbucket_cloud.streams.base import BUCKET_COUNT, STATE_VERSION, repo_state_key
+from source_bitbucket_cloud.streams.branches import BranchesStream
+from source_bitbucket_cloud.streams.commit_branch_reachability import CommitBranchReachabilityStream
+from source_bitbucket_cloud.streams.commits import CommitsStream
+from source_bitbucket_cloud.streams.file_changes import FileChangesStream
+from source_bitbucket_cloud.streams.metric_events import TagsStream
+from tests.conftest import SHARED, FakeCatalog, FakeClient, branch, repository
+
+START_DATE = "2026-01-01"
+WINDOWED = {**SHARED, "start_date": START_DATE}
+RANGE_STREAMS = [CommitsStream, FileChangesStream, CommitBranchReachabilityStream]
+
+
+class CountingClient(FakeClient):
+    def __init__(self):
+        super().__init__()
+        self.branch_calls = 0
+
+    def branches(self, repo):
+        self.branch_calls += 1
+        return self.branch_values.get(repo.uuid, [])
+
+
+def read_all_buckets(stream):
+    records = []
+    for bucket in range(BUCKET_COUNT):
+        records.extend(stream.read_records(None, stream_slice={"bucket_id": bucket}))
+    return records
+
+
+def build(stream_class, client, repo, shared=WINDOWED):
+    stream = stream_class(**{**shared, "client": client, "catalog": FakeCatalog([repo], client)})
+    stream.state = {}
+    return stream
+
+
+@pytest.mark.parametrize("stream_class", RANGE_STREAMS)
+class TestRepositoriesOutsideTheWindow:
+    def test_a_repository_untouched_since_start_date_costs_nothing(self, stream_class):
+        repo = repository(updated_on="2019-05-01T00:00:00+00:00")
+        client = CountingClient()
+        client.branch_values[repo.uuid] = [branch("main", "old")]
+
+        stream = build(stream_class, client, repo)
+        records = read_all_buckets(stream)
+
+        assert records == []
+        assert client.branch_calls == 0, "no push since start_date means nothing to read"
+        assert stream.state["repositories"] == {}, "a gated repository must not grow the state"
+
+    def test_a_repository_pushed_inside_the_window_syncs(self, stream_class):
+        repo = repository(updated_on="2026-06-01T00:00:00+00:00")
+        client = CountingClient()
+        client.branch_values[repo.uuid] = [branch("main", "head", **{"target": {"date": "2026-06-01T00:00:00+00:00"}})]
+        client.commit_values = [{"hash": "head", "date": "2026-06-01T00:00:00+00:00"}]
+
+        stream = build(stream_class, client, repo)
+        read_all_buckets(stream)
+
+        assert client.branch_calls > 0
+
+    def test_a_repository_without_updated_on_is_never_gated(self, stream_class):
+        repo = repository()
+        repo.raw.pop("updated_on")
+        client = CountingClient()
+        client.branch_values[repo.uuid] = [branch("main", "head")]
+        client.commit_values = [{"hash": "head", "date": "2026-06-01T00:00:00+00:00"}]
+
+        stream = build(stream_class, client, repo)
+        read_all_buckets(stream)
+
+        assert client.branch_calls > 0, "an unknown push date must be read, not assumed stale"
+
+    def test_no_start_date_gates_nothing(self, stream_class):
+        repo = repository(updated_on="2019-05-01T00:00:00+00:00")
+        client = CountingClient()
+        client.branch_values[repo.uuid] = [branch("main", "old")]
+        client.commit_values = [{"hash": "old", "date": "2019-05-01T00:00:00+00:00"}]
+
+        stream = build(stream_class, client, repo, shared=SHARED)
+        read_all_buckets(stream)
+
+        assert client.branch_calls > 0
+
+
+class TestBranchesAreCurrentStateNotHistory:
+    """Branches exist now regardless of when they were last pushed to, so the
+    window must not empty a dormant repository's branch snapshot."""
+
+    def dormant(self):
+        repo = repository(updated_on="2019-05-01T00:00:00+00:00")
+        client = CountingClient()
+        client.branch_values[repo.uuid] = [branch("main", "old")]
+        return repo, client
+
+    def test_a_dormant_repository_still_reports_its_branches(self):
+        repo, client = self.dormant()
+
+        records = read_all_buckets(build(BranchesStream, client, repo))
+
+        assert [r["name"] for r in records if r["record_type"] == "item"] == ["main"]
+        marker = next(r for r in records if r["record_type"] == "snapshot_complete")
+        assert marker["snapshot_available"] is True
+        assert marker["snapshot_item_count"] == 1
+
+    def test_and_costs_one_listing_ever(self):
+        repo, client = self.dormant()
+        first = build(BranchesStream, client, repo)
+        read_all_buckets(first)
+
+        second = build(BranchesStream, client, repo)
+        second.state = first.state
+        read_all_buckets(second)
+
+        assert client.branch_calls == 1, "the idle gate, not start_date, is what keeps a dormant repository cheap"
+
+
+def dated_branch(name: str, sha: str, target_date: str | None):
+    return BranchRef(name=name, head_sha=sha, target_date=target_date, is_default=name == "main")
+
+
+class TestColdRepositoriesSkipStaleBranches:
+    """A repository inside the window can still carry branches parked years
+    ago; ranging those on a first read pages their whole history for commits
+    the date filter then discards."""
+
+    def make(self, stream_class):
+        repo = repository(updated_on="2026-06-01T00:00:00+00:00")
+        client = FakeClient()
+        client.branch_values[repo.uuid] = [
+            dated_branch("main", "fresh", "2026-06-01T00:00:00+00:00"),
+            dated_branch("ancient", "stale", "2015-02-01T00:00:00+00:00"),
+        ]
+        client.commit_values = [{"hash": "fresh", "date": "2026-06-01T00:00:00+00:00"}]
+        return build(stream_class, client, repo), client, repo
+
+    @pytest.mark.parametrize("stream_class", [CommitsStream, FileChangesStream])
+    def test_first_read_ranges_only_in_window_heads(self, stream_class):
+        stream, client, _ = self.make(stream_class)
+
+        read_all_buckets(stream)
+
+        assert client.commit_calls == [(["fresh"], [])]
+
+    @pytest.mark.parametrize("stream_class", [CommitsStream, FileChangesStream])
+    def test_stale_heads_are_still_stored_so_a_later_push_diffs(self, stream_class):
+        stream, _, repo = self.make(stream_class)
+
+        read_all_buckets(stream)
+
+        assert stream.state["repositories"][repo_state_key(repo)]["head_shas"] == ["fresh", "stale"]
+
+    def test_reachability_skips_the_stale_branch_only(self):
+        stream, client, _ = self.make(CommitBranchReachabilityStream)
+
+        records = read_all_buckets(stream)
+
+        assert {r["branch_name"] for r in records if r["record_type"] == "item"} == {"main"}
+        assert client.commit_calls == [(["fresh"], [])]
+
+    @pytest.mark.parametrize("stream_class", [CommitsStream, FileChangesStream])
+    def test_a_known_repository_ranges_every_head(self, stream_class):
+        stream, client, repo = self.make(stream_class)
+        stream.state = {
+            "version": STATE_VERSION,
+            "bucket_count": BUCKET_COUNT,
+            "repositories": {repo_state_key(repo): {"head_shas": ["older"], "repo_updated_on": "2026-05-01"}},
+        }
+
+        read_all_buckets(stream)
+
+        assert client.commit_calls == [(["fresh", "stale"], ["older"])], (
+            "once heads are known the exclude set bounds the read, so nothing needs skipping"
+        )
+
+    def test_a_branch_without_a_target_date_is_read(self):
+        repo = repository(updated_on="2026-06-01T00:00:00+00:00")
+        client = FakeClient()
+        client.branch_values[repo.uuid] = [dated_branch("main", "undated", None)]
+        client.commit_values = [{"hash": "undated", "date": "2026-06-01T00:00:00+00:00"}]
+
+        read_all_buckets(build(CommitsStream, client, repo))
+
+        assert client.commit_calls == [(["undated"], [])]
+
+
+class TestTagsHonourTheWindow:
+    def tag(self, name: str, tagged_at: str | None, commit_date: str = "2015-01-01T00:00:00+00:00"):
+        """An annotated tag carries its own date; `target` is the commit it
+        points at, which can be far older than the tag."""
+        record = {"name": name, "target": {"hash": f"{name}-sha", "date": commit_date}}
+        if tagged_at:
+            record["date"] = tagged_at
+        return record
+
+    def read(self, tags):
+        repo = repository(updated_on="2026-06-01T00:00:00+00:00")
+        client = FakeClient()
+        client.optional_values[client.repo_path(repo, "refs/tags")] = (True, tags)
+        stream = TagsStream(**{**WINDOWED, "client": client, "catalog": FakeCatalog([repo], client)})
+        return read_all_buckets(stream)
+
+    def test_tags_older_than_start_date_are_dropped(self):
+        records = self.read([self.tag("v1", "2015-01-01T00:00:00+00:00"), self.tag("v9", "2026-06-01T00:00:00+00:00")])
+
+        assert [r["name"] for r in records if r["record_type"] == "item"] == ["v9"]
+
+    def test_the_marker_counts_only_what_was_emitted(self):
+        records = self.read([self.tag("v1", "2015-01-01T00:00:00+00:00"), self.tag("v9", "2026-06-01T00:00:00+00:00")])
+
+        marker = next(r for r in records if r["record_type"] == "snapshot_complete")
+        assert marker["snapshot_item_count"] == 1, "completeness must match the filtered snapshot"
+        assert marker["snapshot_available"] is True
+
+    def test_an_undated_tag_is_kept(self):
+        records = self.read([self.tag("v1", None)])
+
+        assert [r["name"] for r in records if r["record_type"] == "item"] == ["v1"]
+
+    def test_a_new_tag_on_an_old_commit_is_kept(self):
+        """Judging a tag by its commit would drop every release cut against
+        history — the tag is the event, not the commit it names."""
+        records = self.read([self.tag("v9", "2026-06-01T00:00:00+00:00", commit_date="2015-01-01T00:00:00+00:00")])
+
+        assert [r["name"] for r in records if r["record_type"] == "item"] == ["v9"]
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_uncomputable_diff.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_uncomputable_diff.py
new file mode 100644
index 000000000..337042a60
--- /dev/null
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_uncomputable_diff.py
@@ -0,0 +1,125 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from source_bitbucket_cloud.client import UNCOMPUTABLE_DIFF, BitbucketApiError, BitbucketClient
+from source_bitbucket_cloud.streams.base import BUCKET_COUNT
+from source_bitbucket_cloud.streams.pr_diffstat import PRDiffstatStream
+from tests.conftest import SHARED, FakeCatalog, repository
+
+
+def read_all_buckets(stream):
+    records = []
+    for bucket in range(BUCKET_COUNT):
+        records.extend(stream.read_records(None, stream_slice={"bucket_id": bucket}))
+    return records
+
+
+NO_COMMON_ANCESTOR = json.dumps({"type": "error", "error": {"message": "No common ancestor"}})
+MALFORMED_REQUEST = json.dumps({"type": "error", "error": {"message": "Invalid pagelen"}})
+
+
+class FakeResponse:
+    def __init__(self, status_code: int, url: str, payload, text: str = ""):
+        self.status_code = status_code
+        self.url = url
+        self.text = text
+        self._payload = payload
+
+    def json(self):
+        return self._payload
+
+
+class FakeSession:
+    def __init__(self, routes):
+        self.headers: dict[str, str] = {}
+        self._routes = routes
+        self.urls: list[str] = []
+
+    def request(self, method, url, params=None, data=None, timeout=None):
+        del method, params, data, timeout
+        self.urls.append(url)
+        for fragment, response in self._routes:
+            if fragment in url:
+                return FakeResponse(response[0], url, response[1], response[2])
+        raise AssertionError(f"unrouted request: {url}")
+
+
+def client_with(routes) -> BitbucketClient:
+    client = BitbucketClient("token")
+    client._session = FakeSession(routes)
+    return client
+
+
+def pull_request(pr_id: int = 42):
+    return {
+        "id": pr_id,
+        "updated_on": "2026-06-30T00:00:00+00:00",
+        "created_on": "2026-06-01T00:00:00+00:00",
+        "state": "MERGED",
+        "source": {"branch": {"name": "f"}, "commit": {"hash": "src"}},
+        "destination": {"branch": {"name": "main"}, "commit": {"hash": "dst"}},
+    }
+
+
+class TestClientTolerance:
+    def test_uncomputable_diff_reads_as_unavailable(self):
+        client = client_with([("diffstat", (400, None, NO_COMMON_ANCESTOR))])
+
+        present, entries = client.paginate_optional("x/diffstat", tolerate_messages=UNCOMPUTABLE_DIFF)
+
+        assert present is False
+        assert list(entries) == []
+
+    def test_other_400_still_raises(self):
+        client = client_with([("diffstat", (400, None, MALFORMED_REQUEST))])
+
+        with pytest.raises(BitbucketApiError):
+            client.paginate_optional("x/diffstat", tolerate_messages=UNCOMPUTABLE_DIFF)
+
+    def test_tolerance_is_opt_in(self):
+        client = client_with([("diffstat", (400, None, NO_COMMON_ANCESTOR))])
+
+        with pytest.raises(BitbucketApiError):
+            client.paginate_optional("x/diffstat")
+
+    @pytest.mark.parametrize("body", ["", "<html>gateway</html>", json.dumps({"error": "flat"})])
+    def test_unparseable_bodies_have_no_message(self, body):
+        assert BitbucketApiError(400, "u", body).error_message == "", f"should not parse: {body!r}"
+
+
+class TestDiffstatStreamTolerance:
+    def build(self, diffstat_response):
+        repo = repository()
+        client = client_with(
+            [
+                ("diffstat", diffstat_response),
+                ("pullrequests", (200, {"values": [pull_request()]}, "")),
+            ]
+        )
+        stream = PRDiffstatStream(**{**SHARED, "client": client, "catalog": FakeCatalog([repo], client)})
+        stream.state = {}
+        return stream, repo
+
+    def test_pull_request_without_common_ancestor_marks_the_snapshot_unavailable(self):
+        stream, _ = self.build((400, None, NO_COMMON_ANCESTOR))
+
+        records = read_all_buckets(stream)
+
+        markers = [r for r in records if r["record_type"] == "snapshot_complete"]
+        assert markers and markers[0]["snapshot_available"] is False, (
+            "an undefined diff must read as 'could not look', not as an empty change set"
+        )
+        assert not [r for r in records if r["record_type"] == "item"]
+
+    def test_other_400_still_fails_the_repository(self):
+        stream, _ = self.build((400, None, MALFORMED_REQUEST))
+
+        with pytest.raises(RuntimeError, match="repositories failed"):
+            read_all_buckets(stream)
+
+        assert stream._failed_repositories == ["ws/repo"], (
+            "a 400 we do not recognise is a bug, not a permanent API answer"
+        )
diff --git a/src/ingestion/connectors/git/bitbucket-cloud/tests/test_unresolvable_heads.py b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_unresolvable_heads.py
new file mode 100644
index 000000000..233fbe3a1
--- /dev/null
+++ b/src/ingestion/connectors/git/bitbucket-cloud/tests/test_unresolvable_heads.py
@@ -0,0 +1,215 @@
+from __future__ import annotations
+
+import json
+
+import pytest
+
+from source_bitbucket_cloud.client import BitbucketApiError
+from source_bitbucket_cloud.streams.base import BUCKET_COUNT, repo_state_key
+from source_bitbucket_cloud.streams.commit_branch_reachability import CommitBranchReachabilityStream
+from source_bitbucket_cloud.streams.commits import CommitsStream
+from source_bitbucket_cloud.streams.git_ranges import RANGE_REPAIR_ATTEMPTS
+from tests.conftest import SHARED, FakeCatalog, FakeClient, branch, repository
+
+DATE = "2026-06-01T00:00:00+00:00"
+
+
+def commit_not_found(*shas: str) -> BitbucketApiError:
+    body = json.dumps({"type": "error", "error": {"message": "Commit not found", "data": {"shas": list(shas)}}})
+    return BitbucketApiError(404, "https://api.bitbucket.org/2.0/x/commits", body)
+
+
+class UnresolvableClient(FakeClient):
+    """Rejects any range that still references a sha the API cannot resolve."""
+
+    def __init__(self, unresolvable: set[str]):
+        super().__init__()
+        self.unresolvable = unresolvable
+
+    def commits_between(self, repo, include, exclude):
+        self.commit_calls.append((list(include), list(exclude)))
+        referenced = self.unresolvable.intersection(set(include) | set(exclude))
+        if referenced:
+            raise commit_not_found(*sorted(referenced))
+        return iter([{"hash": sha, "date": DATE} for sha in include])
+
+
+def read_all_buckets(stream):
+    records = []
+    for bucket in range(BUCKET_COUNT):
+        records.extend(stream.read_records(None, stream_slice={"bucket_id": bucket}))
+    return records
+
+
+def build(cls, client, repo):
+    stream = cls(**{**SHARED, "client": client, "catalog": FakeCatalog([repo], client)})
+    stream.state = {}
+    return stream
+
+
+class TestMissingShasArePruned:
+    def test_readable_heads_still_sync(self, repo):
+        client = UnresolvableClient({"ghost"})
+        client.branch_values[repo.uuid] = [branch("main", "alive"), branch("old", "ghost")]
+
+        records = read_all_buckets(build(CommitsStream, client, repo))
+
+        assert [r["hash"] for r in records] == ["alive"], (
+            "one unresolvable head must not cost the repository its other branches"
+        )
+        assert client.commit_calls[-1] == (["alive"], [])
+
+    def test_error_names_the_shas_to_drop(self):
+        assert commit_not_found("a", "b").missing_shas == frozenset({"a", "b"})
+
+    @pytest.mark.parametrize("body", ["", "{}", json.dumps({"error": {"data": {"shas": "nope"}}})])
+    def test_absent_sha_lists_read_as_empty(self, body):
+        assert BitbucketApiError(404, "u", body).missing_shas == frozenset()
+
+    def test_repository_with_no_resolvable_head_yields_nothing(self, repo):
+        client = UnresolvableClient({"ghost"})
+        client.branch_values[repo.uuid] = [branch("old", "ghost")]
+
+        stream = build(CommitsStream, client, repo)
+        records = read_all_buckets(stream)
+
+        assert records == []
+        assert stream._failed_repositories == [], "an unresolvable head is not a sync failure"
+        assert stream.state["repositories"][repo_state_key(repo)]["head_shas"] == []
+
+    def test_an_unread_head_is_not_checkpointed(self, repo):
+        client = UnresolvableClient({"ghost"})
+        client.branch_values[repo.uuid] = [branch("main", "alive"), branch("old", "ghost")]
+
+        stream = build(CommitsStream, client, repo)
+        read_all_buckets(stream)
+
+        stored = stream.state["repositories"][repo_state_key(repo)]
+        assert stored["head_shas"] == ["alive"], "recording a head we could not read would claim it was synced"
+        assert stored["repo_updated_on"] == "", "the cursor must stay open so the head is tried again"
+
+    def test_a_head_that_resolves_later_is_read_without_a_new_push(self, repo):
+        """The 404 may be transient. Nothing else moves in the repository — no
+        push, so no new updated_on — and the head must still be picked up."""
+        client = UnresolvableClient({"ghost"})
+        client.branch_values[repo.uuid] = [branch("main", "alive"), branch("old", "ghost")]
+        first = build(CommitsStream, client, repo)
+        read_all_buckets(first)
+
+        client.unresolvable.clear()
+        client.commit_calls.clear()
+        second = build(CommitsStream, client, repo)
+        second.state = first.state
+        records = read_all_buckets(second)
+
+        assert client.commit_calls, "the idle gate must not close over an unread head"
+        assert "ghost" in [r["hash"] for r in records]
+        assert second.state["repositories"][repo_state_key(repo)]["head_shas"] == ["alive", "ghost"]
+
+    def test_stale_excludes_are_still_dropped_when_no_shas_are_named(self, repo):
+        class BareNotFound(FakeClient):
+            def commits_between(self, repo, include, exclude):
+                self.commit_calls.append((list(include), list(exclude)))
+                if exclude:
+                    raise BitbucketApiError(404, "u", "gone")
+                return iter([{"hash": "new", "date": DATE}])
+
+        client = BareNotFound()
+        client.branch_values[repo.uuid] = [branch("main", "new")]
+        stream = build(CommitsStream, client, repo)
+        stream.state = {
+            "version": 3,
+            "bucket_count": 8,
+            "repositories": {repo_state_key(repo): {"head_shas": ["old"], "repo_updated_on": "stale"}},
+        }
+
+        records = read_all_buckets(stream)
+
+        assert [r["hash"] for r in records] == ["new"]
+        assert client.commit_calls == [(["new"], ["old"]), (["new"], [])]
+
+    def test_repair_keeps_going_while_it_is_getting_somewhere(self, repo):
+        """The API names only the shas it noticed, so a repository with many
+        dead heads needs many rounds. Stopping part-way would leave it failing
+        the same way on every future sync."""
+        heads = 20
+
+        class OneAtATime(FakeClient):
+            """Names a single dead head per answer."""
+
+            def commits_between(self, repo, include, exclude):
+                self.commit_calls.append((list(include), list(exclude)))
+                raise commit_not_found(sorted(include)[0])
+
+        client = OneAtATime()
+        client.branch_values[repo.uuid] = [branch(f"b{index}", f"sha{index:02d}") for index in range(heads)]
+        stream = build(CommitsStream, client, repo)
+
+        records = read_all_buckets(stream)
+
+        assert records == []
+        assert stream._failed_repositories == [], "pruning to nothing is not a failure"
+        assert not stream._catalog.is_inaccessible(repo), "nor a denial: the listing was readable"
+        assert len(client.commit_calls) == heads, (
+            f"every dead head must be pruned; stopped after {len(client.commit_calls)} of {heads}"
+        )
+        assert len(client.commit_calls) > RANGE_REPAIR_ATTEMPTS, "and past the warning threshold"
+
+    def test_repair_still_terminates_when_nothing_can_be_pruned(self, repo):
+        class NamesSomethingElse(FakeClient):
+            def commits_between(self, repo, include, exclude):
+                self.commit_calls.append((list(include), list(exclude)))
+                raise commit_not_found("a-sha-not-in-this-range")
+
+        client = NamesSomethingElse()
+        client.branch_values[repo.uuid] = [branch("main", "head")]
+        stream = build(CommitsStream, client, repo)
+        stream.state = {
+            "version": 3,
+            "bucket_count": 8,
+            "repositories": {repo_state_key(repo): {"head_shas": ["old"], "repo_updated_on": "stale"}},
+        }
+
+        read_all_buckets(stream)
+
+        assert len(client.commit_calls) == 2, "clear the excludes once, then give up"
+        assert stream._catalog.is_inaccessible(repo)
+
+    def test_unnamed_404_without_excludes_is_a_denial(self, repo):
+        class BareNotFound(FakeClient):
+            def commits_between(self, repo, include, exclude):
+                self.commit_calls.append((list(include), list(exclude)))
+                raise BitbucketApiError(404, "u", "gone")
+
+        client = BareNotFound()
+        client.branch_values[repo.uuid] = [branch("main", "head")]
+        stream = build(CommitsStream, client, repo)
+
+        read_all_buckets(stream)
+
+        assert stream._failed_repositories == []
+        assert stream._catalog.is_inaccessible(repo)
+
+
+class TestReachabilitySkipsVanishedHeads:
+    def test_vanished_head_skips_its_branch_only(self, repo):
+        client = UnresolvableClient({"ghost"})
+        client.branch_values[repo.uuid] = [branch("main", "alive"), branch("old", "ghost")]
+
+        stream = build(CommitBranchReachabilityStream, client, repo)
+        records = read_all_buckets(stream)
+
+        assert stream._failed_repositories == []
+        branches_seen = {r["branch_name"] for r in records if r["record_type"] == "item"}
+        assert branches_seen == {"main"}
+
+    def test_the_skipped_branch_is_not_checkpointed(self, repo):
+        client = UnresolvableClient({"ghost"})
+        client.branch_values[repo.uuid] = [branch("main", "alive"), branch("old", "ghost")]
+
+        stream = build(CommitBranchReachabilityStream, client, repo)
+        read_all_buckets(stream)
+
+        stored = stream.state["repositories"][repo_state_key(repo)]
+        assert stored["heads"] == {"main": "alive"}
+        assert stored["repo_updated_on"] == ""