Fix/148 resolution sets one commit

Makefile

@@ -124,7 +124,7 @@ orchestration: nightly-worker-job nightly-manager-job compress_buckets
 
 metadata: tag-questions validate-questions
 
-resolve: resolve-forecasts
+resolve: resolve-forecasts push-resolution-sets
 
 leaderboards: leaderboard-tournament leaderboard-baseline leaderboard-preliminary
 
@@ -222,6 +222,9 @@ validate-questions:
 resolve-forecasts:
 	$(MAKE) -C src/orchestration/func_resolve || echo "* $@" >> $(MAKE_FAILURE_LOG)
 
+push-resolution-sets:
+	$(MAKE) -C src/orchestration/func_push_resolution_sets || echo "* $@" >> $(MAKE_FAILURE_LOG)
+
 naive-and-dummy-forecasters:
 	$(MAKE) -C src/base_eval/naive_and_dummy_forecasters || echo "* $@" >> $(MAKE_FAILURE_LOG)
 

src/helpers/git.py

@@ -57,7 +57,7 @@ def clone_and_push_files(
     files: Dict[str, str],
     commit_message: str,
     mirrors: Optional[List[str]] = None,
-) -> None:
+) -> bool:
     """Clone a Git repository, add/update files, commit, and push to origin and optional mirrors.
 
     Args:
@@ -68,7 +68,9 @@ def clone_and_push_files(
                                        If None, attempts to load from secrets.
 
     Returns:
-        None. Exits with status 1 if an error is encountered while pushing.
+        bool: True if a new commit was created, False if HEAD was unchanged.
+              Origin and mirrors are pushed either way. Exits with status 1 if an
+              error is encountered while pushing.
     """
     if not mirrors:
         mirrors = keys.get_secret_that_may_not_exist("HUGGING_FACE_REPO_URL")
@@ -84,19 +86,26 @@ def clone_and_push_files(
         shutil.copy(source, full_destination_path, follow_symlinks=False)
         repo.index.add([destination])
 
+    # Skip empty commits, but always push so a lagging mirror catches up.
+    has_changes = bool(repo.index.diff(repo.head.commit))
+    if not has_changes:
+        logger.info(f"No new commit for {repo_url}; syncing remotes to HEAD.")
+
     error_encountered = False
     author = Actor("ForecastBench bot", constants.BENCHMARK_EMAIL)
     committer = Actor("ForecastBench bot", constants.BENCHMARK_EMAIL)
     ssh_env = {"GIT_SSH_COMMAND": f"ssh -i {tmp_key_file_path} -o StrictHostKeyChecking=no"}
     try:
-        repo.index.commit(commit_message, author=author, committer=committer)
+        if has_changes:
+            repo.index.commit(commit_message, author=author, committer=committer)
         origin = repo.remote(name="origin")
-        origin.push(env=ssh_env)
+        # A rejected push does not raise in GitPython; surface it explicitly.
+        origin.push(env=ssh_env).raise_if_error()
         for index, mirror_url in enumerate(mirrors):
             mirror = repo.create_remote(f"mirror_{index}", url=mirror_url)
-            mirror.push(env=ssh_env)
+            mirror.push(env=ssh_env).raise_if_error()
             repo.delete_remote(mirror.name)
-            logger.info(f"Pushed to {mirror_url} (mirror) with commit message: {commit_message}")
+            logger.info(f"Pushed to {mirror_url} (mirror)")
     except Exception as e:
         error_encountered = True
         message = e.message if hasattr(e, "message") else str(e)
@@ -108,14 +117,16 @@ def clone_and_push_files(
     if error_encountered:
         sys.exit(1)
 
-    logger.info(f"Pushed to {repo_url} with commit message: {commit_message}")
+    if has_changes:
+        logger.info(f"Pushed to {repo_url} with commit message: {commit_message}")
+    return has_changes
 
 
 def clone_commit_and_push(
     files: Dict[str, str],
     commit_message: str,
 ) -> None:
-    """Upload files files to Cloud Storage and push updates to Git.
+    """Push files to the git dataset repository.
 
     Args:
         files (Dict[str, str]): Mapping of local file paths to their git location.

src/nightly_update_workflow/manager/main.py

@@ -181,14 +181,27 @@ def main():
         dict_to_use=dict_to_use_naive_and_dummy_forecasters, task_count=1
     )
 
-    # Block on resolve forecasts before launching leaderboards
+    # Block on resolve forecasts before pushing resolution sets to git
     cloud_run.block_and_check_job_result(
         operation=operation_resolve_forecasts,
         name=dict_to_use_resolve_forecasts,
         exit_on_error=True,
         timeout=timeout_resolve_forecasts,
     )
 
+    # Push all resolution sets in a single dedicated job now that the parallel resolve tasks have
+    # finished, so only one process pushes to the dataset repo (avoids a race condition).
+    dict_to_use_push_resolution_sets = "push_resolution_sets"
+    operation_push_resolution_sets = call_worker(
+        dict_to_use=dict_to_use_push_resolution_sets,
+        task_count=1,
+    )
+    cloud_run.block_and_check_job_result(
+        operation=operation_push_resolution_sets,
+        name=dict_to_use_push_resolution_sets,
+        exit_on_error=False,
+    )
+
     # Launch leaderboard jobs in parallel
     dict_to_use_leaderboards = "leaderboards"
     timeout_leaderboards = cloud_run.timeout_1h * 4

src/nightly_update_workflow/worker/main.py

@@ -17,6 +17,15 @@
         ("func-resolve-forecasts", True, cloud_run.timeout_1h * 3, 50),
     ]
 ]
+
+
+push_resolution_sets = [
+    [
+        ("func-push-resolution-sets", True, cloud_run.timeout_1h, 1),
+    ]
+]
+
+
 leaderboards = [
     [
         ("func-leaderboard-tournament", True, cloud_run.timeout_1h * 4, 1),
@@ -139,14 +148,16 @@ def main():
 
     Env variables:
     CLOUD_RUN_TASK_INDEX: automatically set by Cloud Run Jobs
-    DICT_TO_USE: one of `fetch_and_update`, `metadata`, `resolve_forecasts`, `leaderboards`.
+    DICT_TO_USE: one of `fetch_and_update`, `metadata`, `resolve_forecasts`,
+                 `push_resolution_sets`, `leaderboards`.
     """
     dict_mapping = {
         "fetch_and_update": get_fetch_and_update(),
         "metadata": metadata,
         "create_question_set": get_create_question_set(),
         "publish_question_set_make_llm_baseline": get_publish_question_set_make_llm_baseline(),
         "resolve_forecasts": resolve_forecasts,
+        "push_resolution_sets": push_resolution_sets,
         "leaderboards": leaderboards,
         "naive_and_dummy_forecasters": get_naive_and_dummy_forecasters(),
         "website": website,

src/orchestration/_io.py

@@ -233,10 +233,11 @@ def upload_hash_mapping(raw_json: str, source_name: str) -> None:
 
 
 def upload_resolution_set(df: pd.DataFrame, forecast_due_date: str, question_set_filename: str):
-    """Upload resolution set to GCS and push to git."""
-    from helpers import git  # noqa: E402
-    from helpers import keys  # noqa: E402
+    """Upload resolution set to GCS.
 
+    Only uploads to the bucket; resolution sets are pushed to git later by
+    `push_all_resolution_sets`.
+    """
     basename = f"{forecast_due_date}_resolution_set.json"
     local_filename = f"/tmp/{basename}"
     df = df[["id", "source", "direction", "resolution_date", "resolved_to", "resolved"]]
@@ -259,14 +260,52 @@ def upload_resolution_set(df: pd.DataFrame, forecast_due_date: str, question_set
     )
     logger.info(f"Uploaded Resolution File {local_filename} to {upload_folder}.")
 
+
+def push_all_resolution_sets() -> None:
+    """Push every resolution set in the bucket to the git dataset repo in a single commit.
+
+    Run as one job rather than having each parallel resolve task push its own set, which
+    avoids the race condition of concurrent pushes to the repository.
+    """
+    from helpers import git, keys  # noqa: E402
+
+    if env.RUNNING_LOCALLY:
+        logger.info("Running locally; not pushing resolution sets to git.")
+        return
+
+    folder = "datasets/resolution_sets"
+    blob_names = gcp.storage.list_with_prefix(
+        bucket_name=env.PUBLIC_RELEASE_BUCKET,
+        prefix=f"{folder}/",
+    )
+
+    files = {}
+    for blob_name in blob_names:
+        basename = os.path.basename(blob_name)
+        if not basename.endswith("_resolution_set.json"):
+            continue
+        local_filename = f"/tmp/{basename}"
+        gcp.storage.download(
+            bucket_name=env.PUBLIC_RELEASE_BUCKET,
+            filename=blob_name,
+            local_filename=local_filename,
+        )
+        files[local_filename] = f"{folder}/{basename}"
+
+    if not files:
+        logger.warning(f"No resolution sets found under {folder}; nothing to push.")
+        return
+
     mirrors = keys.get_secret_that_may_not_exist("HUGGING_FACE_REPO_URL")
     mirrors = [mirrors] if mirrors else []
-    git.clone_and_push_files(
+    committed = git.clone_and_push_files(
         repo_url=keys.API_GITHUB_DATASET_REPO_URL,
-        files={local_filename: f"{upload_folder}/{basename}"},
-        commit_message=f"resolution set: automatic update for {question_set_filename}.",
+        files=files,
+        commit_message="resolution sets: automatic update.",
         mirrors=mirrors,
     )
+    if committed:
+        logger.info(f"Pushed {len(files)} resolution sets to git in a single commit.")
 
 
 def upload_processed_forecast_file(data: dict, forecast_due_date: str, filename: str):

src/orchestration/func_push_resolution_sets/Makefile

@@ -0,0 +1,37 @@
+all :
+	$(MAKE) clean
+	$(MAKE) deploy
+
+.PHONY : all clean deploy
+
+UPLOAD_DIR = upload
+ROOT_DIR ?= $(abspath ../../..)/
+include $(ROOT_DIR)orchestration_upload.mk
+
+.gcloudignore:
+	cp -r $(ROOT_DIR)src/helpers/.gcloudignore .
+
+Dockerfile: $(ROOT_DIR)src/helpers/Dockerfile.template
+	sed \
+		-e 's/REGION/$(CLOUD_DEPLOY_REGION)/g' \
+		-e 's/STACK/google-22-full/g' \
+		-e 's/PYTHON_VERSION/python312/g' \
+		$< > Dockerfile
+
+deploy : main.py .gcloudignore requirements.txt Dockerfile
+	$(stage-orchestration-upload)
+	gcloud run jobs deploy \
+		func-push-resolution-sets \
+		--project $(CLOUD_PROJECT) \
+		--region $(CLOUD_DEPLOY_REGION) \
+		--tasks 1 \
+		--parallelism 1 \
+		--task-timeout 1h \
+		--memory 2Gi \
+		--max-retries 0 \
+		--service-account $(QUESTION_BANK_BUCKET_SERVICE_ACCOUNT) \
+		--set-env-vars $(DEFAULT_CLOUD_FUNCTION_ENV_VARS) \
+		--source $(UPLOAD_DIR)
+
+clean :
+	rm -rf $(UPLOAD_DIR) .gcloudignore Dockerfile

src/orchestration/func_push_resolution_sets/main.py

@@ -0,0 +1,23 @@
+"""Cloud Run job: push all resolution sets to the git dataset repo in a single commit.
+
+See `orchestration._io.push_all_resolution_sets`.
+"""
+
+import logging
+from typing import Any
+
+from helpers import decorator
+from orchestration import _io
+
+logging.basicConfig(level=logging.INFO)
+logger = logging.getLogger(__name__)
+
+
+@decorator.log_runtime
+def driver(_: Any) -> None:
+    """Push all resolution sets to git in a single commit."""
+    _io.push_all_resolution_sets()
+
+
+if __name__ == "__main__":
+    driver(None)

src/orchestration/func_push_resolution_sets/requirements.txt

@@ -0,0 +1,10 @@
+google-cloud-storage
+google-cloud-secret-manager
+pandas>=2.2.2,<3.0
+GitPython
+termcolor
+slack_sdk
+pandera
+pytz
+python-dateutil
+backoff