[DevOps]: Add Chrysalis ingestion wrapper and deploy

backend/app/scripts/README.md

@@ -25,7 +25,9 @@ scripts/
 │   ├── sites/
 │   │   ├── lcrc-diagnostics-scanner.sh
 │   │   ├── nersc-diagnostics-scanner.sh
-│   │   └── nersc.sh
+│   │   ├── site_ingestion_launcher.sh
+│   │   ├── chrysalis.config
+│   │   └── nersc.config
 │   └── v3_data/
 │       ├── __init__.py
 │       ├── lcrc-v3.env.example
@@ -58,6 +60,7 @@ Example:
 python -m app.scripts.db.seed
 python -m app.scripts.db.rollback_seed
 python -m app.scripts.users.create_admin_account
+python -m app.scripts.ingestion.hpc_upload_archive_ingestor
 python -m app.scripts.ingestion.nersc_archive_ingestor
 python -m app.scripts.ingestion.v3_data.lcrc_v3_archive_ingestor
 ```
@@ -120,6 +123,42 @@ If operational complexity increases, these scripts may later be consolidated int
 
 ---
 
+## HPC Upload Archive Ingestor
+
+The scheduler-agnostic HPC upload archive ingestor is the preferred entrypoint for
+site wrappers. It currently delegates to the existing NERSC archive ingestor,
+preserving Perlmutter behavior while giving non-NERSC schedulers a stable shared
+command.
+
+Example:
+
+```bash
+uv run python -m app.scripts.ingestion.hpc_upload_archive_ingestor
+```
+
+### Site Collection Launcher
+
+`app/scripts/ingestion/sites/site_ingestion_launcher.sh` is the host-side
+launcher for site collection. It loads `sites/<site>.config`, then selects the
+configured Python ingestor. Use it as:
+
+```bash
+app/scripts/ingestion/sites/site_ingestion_launcher.sh nersc staging
+app/scripts/ingestion/sites/site_ingestion_launcher.sh chrysalis archive
+```
+
+Each site config defines its machine name, archive roots, working and repository
+paths, Python environment file, token export file, API base URL, archive lower
+bound, and ingestor module. The launcher defaults to `DRY_RUN=true` with
+`DRY_RUN_USE_REMOTE_STATE=true`, so it loads API credentials and performs
+read-only state validation. Set `DRY_RUN_USE_REMOTE_STATE=false` for a
+credential-free offline scan. Set `DRY_RUN=false` only after validating archive
+access, token storage, network egress, and candidate counts. A capped
+`MAX_CASES_PER_RUN` value limits real ingestion but still persists results.
+
+Site configs are operational inputs. Keep credentials in their referenced,
+protected files rather than committing them to a config file.
+
 ## NERSC Archive Ingestor
 
 The NERSC archive ingestor scans a bind-mounted performance archive directory,
@@ -148,23 +187,19 @@ Configuration surface (via env vars):
 - `OLD_PERF_ARCHIVE_ROOT` (default `/OLD_PERF` for `SCAN_MODE=archive`)
 - `MACHINE_NAME` (default `perlmutter`)
 - `DRY_RUN` (default `true`)
+- `DRY_RUN_USE_REMOTE_STATE` (default `true`; set `false` for offline dry runs)
 - `MAX_CASES_PER_RUN` (optional, default not set)
 - `MAX_ATTEMPTS` (optional, default not set)
 - `REQUEST_TIMEOUT_SECONDS` (optional, default 60)
 - `ARCHIVE_YEAR_START` (optional, archive mode only; accepts `YYYY` or `YYYY-MM`)
 - `ARCHIVE_YEAR_END` (optional, archive mode only; accepts `YYYY` or `YYYY-MM`)
 
-Helper wrapper:
-
-- `backend/app/scripts/ingestion/sites/nersc.sh` activates `backend/.venv`, sets the documented NERSC staging and archive roots, defaults to `SCAN_MODE=archive`, defaults to `DRY_RUN=true`, and then runs `python -m app.scripts.ingestion.nersc_archive_ingestor`.
-- Override `SCAN_MODE`, `DRY_RUN`, or any other supported env var in the caller or cron entry when you need a different schedule or behavior.
-
 Archive notes:
 
 - Archive mode traverses only top-level `YYYY-MM` directories under `OLD_PERF_ARCHIVE_ROOT`. Other top-level directories are ignored.
 - Archive scans may include paths without a `COMPLETED/` directory. When snapshot status buckets exist, ingestor scans only `COMPLETED/` and ignores sibling directories in that snapshot bucket.
 - Archive dedupe is based on logical case identity plus `execution_id`, not the full timestamped snapshot path.
-- `ARCHIVE_YEAR_START` / `ARCHIVE_YEAR_END` are intended for scoped backfills so operators can avoid scanning the full historical tree when unnecessary.
+- Direct Python entrypoints leave `ARCHIVE_YEAR_START` / `ARCHIVE_YEAR_END` unset. The site collection launcher applies each site's configured archive lower bound; callers may override either bound for a differently scoped archive scan.
 - `YYYY` values expand to full-year bounds (`START=2020` means `2020-01`; `END=2020` means `2020-12`), while `YYYY-MM` values target exact archive month buckets.
 
 ## One-Time Chrysalis E3SM v3 Archive Backfill
@@ -288,6 +323,7 @@ Configuration surface (via env vars):
 - `OLD_PERF_ARCHIVE_ROOT` (default `/OLD_PERF` for `SCAN_MODE=archive`)
 - `MACHINE_NAME` (default `perlmutter`)
 - `DRY_RUN` (default `true`)
+- `DRY_RUN_USE_REMOTE_STATE` (default `true`; set `false` for offline dry runs)
 - `MAX_CASES_PER_RUN` (optional, default not set)
 - `MAX_ATTEMPTS` (optional, default not set)
 - `REQUEST_TIMEOUT_SECONDS` (optional, default 60)

backend/app/scripts/ingestion/archive_ingestor_core.py

@@ -132,6 +132,7 @@
     "startup_configuration_runtime": (
         "machine_name",
         "dry_run",
+        "dry_run_use_remote_state",
         "max_cases_per_run",
         "max_attempts",
         "request_timeout_seconds",
@@ -261,6 +262,8 @@ class IngestorConfig:
     archive_year_start: str | None = None
     # Optional archive upper bound normalized to a YYYY-MM archive bucket.
     archive_year_end: str | None = None
+    # Whether a dry run reads existing state and archive checkpoints from SimBoard.
+    dry_run_use_remote_state: bool = True
 
 
 class IngestionRequestError(Exception):
@@ -504,6 +507,9 @@ def _build_config_from_env(
 
     machine_name = os.getenv("MACHINE_NAME", DEFAULT_MACHINE_NAME)
     dry_run = _parse_bool(os.getenv("DRY_RUN"), default=True)
+    dry_run_use_remote_state = _parse_bool(
+        os.getenv("DRY_RUN_USE_REMOTE_STATE"), default=True
+    )
     max_cases_per_run = _parse_optional_int(os.getenv("MAX_CASES_PER_RUN"))
 
     if max_cases_per_run is not None and max_cases_per_run <= 0:
@@ -557,6 +563,7 @@ def _build_config_from_env(
         machine_name=machine_name,
         scan_mode=cast(Literal["staging", "archive"], scan_mode),
         dry_run=dry_run,
+        dry_run_use_remote_state=dry_run_use_remote_state,
         max_cases_per_run=max_cases_per_run,
         max_attempts=max_attempts,
         request_timeout_seconds=timeout_seconds,

backend/app/scripts/ingestion/archive_workflow.py

@@ -48,7 +48,7 @@ def _validate_run_preconditions(
         )
         return False
 
-    if not config.api_token:
+    if (not config.dry_run or config.dry_run_use_remote_state) and not config.api_token:
         log_event_fn(
             "configuration_error",
             {"error": "SIMBOARD_API_TOKEN is required"},
@@ -90,6 +90,7 @@ def _log_startup_configuration(
         {
             "machine_name": config.machine_name,
             "dry_run": config.dry_run,
+            "dry_run_use_remote_state": config.dry_run_use_remote_state,
             "max_cases_per_run": config.max_cases_per_run,
             "max_attempts": config.max_attempts,
             "request_timeout_seconds": config.request_timeout_seconds,

backend/app/scripts/ingestion/hpc_upload_archive_ingestor.py

@@ -5,16 +5,18 @@
 is read from environment variables (for example ``SIMBOARD_API_BASE_URL``,
 ``SIMBOARD_API_TOKEN``, ``PERF_ARCHIVE_ROOT``, ``OLD_PERF_ARCHIVE_ROOT``, and ``DRY_RUN``).
 
-Each ingest run executes these phases:
+Non-dry-run ingestion executes these phases:
 
     1. In archive mode, fetch completed snapshot checkpoints.
     2. Fetch persisted per-case state from SimBoard API.
     3. Discover and collect parseable execution directories grouped by case path.
     4. Persist discovery results, then package and submit each changed case.
     5. In archive mode, settle and persist completed snapshot checkpoints.
 
-Dry runs stop after discovery and emit a summary. Successful ingestions update
-database state used to keep future runs idempotent.
+Dry runs read remote state and checkpoints by default, stop after discovery,
+and emit a summary without writes. Set ``DRY_RUN_USE_REMOTE_STATE=false`` for
+an offline dry run with empty local state. Successful ingestions update database
+state used to keep future runs idempotent.
 
 Structured log metric definitions for this runner live in
 ``docs/architecture/metadata-ingestion.md``. This module emits those field names
@@ -33,7 +35,7 @@
 from collections.abc import Iterator
 from contextlib import contextmanager
 from pathlib import Path
-from typing import Callable
+from typing import Any, Callable
 
 from app.features.ingestion.parsers.parser import _locate_metadata_files
 from app.scripts.ingestion.archive_client import (
@@ -63,6 +65,7 @@
     MetadataLocator,
     SleepCallback,
     _build_config_from_env,
+    _fresh_state,
     _log_event,
 )
 from app.scripts.ingestion.archive_workflow import (
@@ -117,21 +120,13 @@ def _case_submission_callback(
     )
 
 
-def _run_ingestor(
+def _prepare_run_state(
     config: IngestorConfig,
-    metadata_locator: MetadataLocator = _locate_metadata_files,
-    sleep_fn: SleepCallback = time.sleep,
-    post_request_fn: CaseSubmissionCallback | None = None,
-    discovery_post_request_fn: DiscoveryResultsPersistenceCallback | None = None,
-    checkpoint_post_request_fn: ArchiveCheckpointPersistenceCallback | None = None,
-    case_path_filter: Callable[[Path], bool] | None = None,
-    additional_dir_pruner: Callable[[str, list[str]], None] | None = None,
     archive_checkpointing: bool = True,
-    run_report: IngestorRunReport | None = None,
-) -> int:
-    """Execute one complete archive scan-and-upload cycle."""
-    use_prepared_archives = post_request_fn is None
-    post_request_fn = _case_submission_callback(post_request_fn)
+) -> tuple[dict[str, Any], set[str], str] | None:
+    """Build offline dry-run state or fetch state needed for this run."""
+    if config.dry_run and not config.dry_run_use_remote_state:
+        return _fresh_state(), set(), ""
 
     endpoint_url = _build_endpoint_url(config)
     state_endpoint_url = _build_state_endpoint_url(config)
@@ -141,10 +136,6 @@ def _run_ingestor(
         state_endpoint_url=state_endpoint_url,
         log_event_fn=_log_event,
     )
-
-    if not _validate_run_preconditions(config, log_event_fn=_log_event):
-        return 1
-
     completed_snapshot_keys: set[str] = set()
     if config.scan_mode == "archive" and archive_checkpointing:
         try:
@@ -162,7 +153,7 @@ def _run_ingestor(
                 "archive_checkpoint_fetch_failed",
                 {"status_code": exc.status_code, "error": str(exc)},
             )
-            return 1
+            return None
 
     try:
         state = _fetch_ingestion_state(
@@ -180,7 +171,37 @@ def _run_ingestor(
                 "error": str(exc),
             },
         )
+        return None
+
+    return state, completed_snapshot_keys, endpoint_url
+
+
+def _run_ingestor(
+    config: IngestorConfig,
+    metadata_locator: MetadataLocator = _locate_metadata_files,
+    sleep_fn: SleepCallback = time.sleep,
+    post_request_fn: CaseSubmissionCallback | None = None,
+    discovery_post_request_fn: DiscoveryResultsPersistenceCallback | None = None,
+    checkpoint_post_request_fn: ArchiveCheckpointPersistenceCallback | None = None,
+    case_path_filter: Callable[[Path], bool] | None = None,
+    additional_dir_pruner: Callable[[str, list[str]], None] | None = None,
+    archive_checkpointing: bool = True,
+    run_report: IngestorRunReport | None = None,
+) -> int:
+    """Execute one complete archive scan-and-upload cycle."""
+    use_prepared_archives = post_request_fn is None
+    post_request_fn = _case_submission_callback(post_request_fn)
+
+    if not _validate_run_preconditions(config, log_event_fn=_log_event):
+        return 1
+
+    run_state = _prepare_run_state(
+        config,
+        archive_checkpointing=archive_checkpointing,
+    )
+    if run_state is None:
         return 1
+    state, completed_snapshot_keys, endpoint_url = run_state
 
     new_discovery_results: list[ExecutionDiscoveryResult] = []
     try: