diff --git a/.github/workflows/docker-build.yml b/.github/workflows/docker-build.yml
index 4b3b67b..f3a67a2 100644
--- a/.github/workflows/docker-build.yml
+++ b/.github/workflows/docker-build.yml
@@ -14,12 +14,26 @@ on:
         type: boolean
 
 jobs:
+  test:
+    runs-on: ubuntu-24.04
+
+    steps:
+    - uses: actions/checkout@v7
+    - uses: actions/setup-python@v7
+      with:
+        python-version: '3.12'
+    - name: Install dependencies
+      run: pip install pytest cwltool==3.1.20250110105449 cwl_utils==0.35 PyYAML pystac==1.15.2 maap-py requests "urllib3>=2"
+    - name: Run tests
+      run: pytest tests
+
   build_and_push:
+    needs: test
 
-    runs-on: ubuntu-latest
+    runs-on: ubuntu-24.04
 
     steps:
-    - uses: actions/checkout@v4
+    - uses: actions/checkout@v7
     - name: Set repo name and tag
       id: vars
       run: |
@@ -42,24 +56,24 @@ jobs:
         echo "tag=$TAG_NAME" >> "$GITHUB_OUTPUT"
 
     - name: Log in to GitHub Container Registry
-      uses: docker/login-action@v2
+      uses: docker/login-action@v4
       with:
         registry: ghcr.io
         username: ${{ github.actor }}
         password: ${{ secrets.GITHUB_TOKEN }}
 
     - name: Log in to Private Registry
-      uses: docker/login-action@v2
+      uses: docker/login-action@v4
       with:
         registry: ${{ secrets.PRIVATE_REGISTRY_URL }}
         username: ${{ secrets.PRIVATE_REGISTRY_USERNAME }}
         password: ${{ secrets.PRIVATE_REGISTRY_PASSWORD }}
 
     - name: Set up Docker Buildx
-      uses: docker/setup-buildx-action@v3
+      uses: docker/setup-buildx-action@v4
 
     - name: Build and push
-      uses: docker/build-push-action@v6
+      uses: docker/build-push-action@v7
       with:
         platforms: linux/amd64
         push: true
diff --git a/create_inputs.py b/create_inputs.py
index 839e6d1..b2c46ba 100644
--- a/create_inputs.py
+++ b/create_inputs.py
@@ -1,5 +1,6 @@
 import json
 import pathlib
+import time
 
 import yaml
 import os
@@ -93,6 +94,9 @@ def convert_value_to_cwl_type(value, cwl_type_str):
         return value
 
 
+HTTP_RETRIES = 4
+
+
 def load_json_from_uri(uri):
     parsed = urllib.parse.urlparse(uri)
 
@@ -110,17 +114,20 @@ def load_json_from_uri(uri):
             print(f"Failed to parse JSON: {e}")
             return None
 
-    # http / https
-    try:
-        response = requests.get(uri)
-        response.raise_for_status()
-        return response.json()
-    except requests.exceptions.RequestException as e:
-        print(f"Request error: {e}")
-        return None
-    except ValueError as e:
-        print(f"Failed to parse JSON: {e}")
-        return None
+    # http / https: retry transient failures (e.g. CMR STAC 503s)
+    for attempt in range(1, HTTP_RETRIES + 1):
+        try:
+            response = requests.get(uri, timeout=60)
+            response.raise_for_status()
+            return response.json()
+        except requests.exceptions.RequestException as e:
+            print(f"Request error (attempt {attempt}/{HTTP_RETRIES}): {e}")
+            if attempt < HTTP_RETRIES:
+                time.sleep(2 ** attempt)
+        except ValueError as e:
+            print(f"Failed to parse JSON: {e}")
+            return None
+    return None
 
 
 def parse_cwl_inputs(cwl_file_path) -> [CommandInputParameter]:
@@ -195,6 +202,15 @@ def map_inputs_from_context(job_inputs: dict, cwl_inputs: [CommandInputParameter
         input_name = get_id_from_uri(input.id)
         base_type, is_optional = parse_cwl_type(input_type)
 
+        if base_type in ("Directory", "File"):
+            raw_value = job_inputs.get(input_name)
+            if raw_value is None or raw_value == "None" or raw_value == "":
+                if is_optional:
+                    # Omit so cwltool does not pass the flag to the algorithm
+                    print(f"No value provided for optional {base_type} input '{input_name}', skipping.")
+                    continue
+                raise ValueError(f"No value provided for required {base_type} input '{input_name}'.")
+
         if base_type == "Directory":
             print("Encountered STAC input, localizing assets.")
             # Load STAC JSON directly from the URI rather than using the HySDS-localized copy.
@@ -202,7 +218,10 @@ def map_inputs_from_context(job_inputs: dict, cwl_inputs: [CommandInputParameter
             # URI, HySDS serves the stale cached version. Fetching from the URI at runtime
             # guarantees we always get the current content.
             # TODO: Update the HySDS spec builder to not map this input as a localize field.
-            stac_json = load_json_from_uri(job_inputs.get(input_name))
+            stac_uri = job_inputs.get(input_name)
+            stac_json = load_json_from_uri(stac_uri)
+            if stac_json is None:
+                raise RuntimeError(f"Could not load STAC JSON for input '{input_name}' from {stac_uri}")
             output_dir = str(pathlib.Path(input_name).resolve())
             stage_in.download_stac_assets_from_json(stac_json, output_dir)
             workflow_inputs[input_name] = {"class": "Directory", "path": output_dir}
diff --git a/docker/Dockerfile b/docker/Dockerfile
index 3b41ce6..aded5a2 100644
--- a/docker/Dockerfile
+++ b/docker/Dockerfile
@@ -21,9 +21,10 @@ RUN pip install --no-cache-dir \
     cwl_utils==0.35 \
     PyYAML \
     awscli \
-    pystac \
-    maap-py \
-    requests
+    pystac==1.15.2 \
+    maap-py==5.1.0 \
+    requests \
+    "urllib3>=2"
 
 # Copy only what you need
 COPY docker_stats/docker-stats-on-exit-shim /docker-stats-on-exit-shim
diff --git a/dps_wrapper.sh b/dps_wrapper.sh
index 72edadb..63bda48 100644
--- a/dps_wrapper.sh
+++ b/dps_wrapper.sh
@@ -4,7 +4,7 @@ USERNAME=$(python /app/get_username.py)
 source /maap-dps.env
 export MAAP_API_HOST
 export DPS_MACHINE_TOKEN
-MAAP_PGT=$(python /app/get_maap_pgt_token.py)
+MAAP_PGT=$(python /app/get_maap_pgt_token.py) || exit $?
 export MAAP_PGT
 unset DPS_MACHINE_TOKEN
 
diff --git a/environment.yaml b/environment.yaml
index 352cd0a..0e1c907 100644
--- a/environment.yaml
+++ b/environment.yaml
@@ -3,5 +3,5 @@ dependencies:
   - cwltool==3.1.20250110105449
   - cwl_utils==0.35
   - nodejs
-  - pystac
-  - maap-py
\ No newline at end of file
+  - pystac==1.15.2
+  - maap-py==5.1.0
\ No newline at end of file
diff --git a/get_maap_pgt_token.py b/get_maap_pgt_token.py
index 28d3546..6334ea8 100644
--- a/get_maap_pgt_token.py
+++ b/get_maap_pgt_token.py
@@ -1,24 +1,104 @@
 import json
 import os
+import sys
+
 import requests
+from requests.adapters import HTTPAdapter
+from urllib3.util.retry import Retry  # backoff_jitter requires urllib3>=2
+
+TIMEOUT = (5, 30)  # (connect, read) seconds
+MAX_RETRIES = 5  # i.e. up to 6 attempts
+BACKOFF_FACTOR = 1.0  # sleeps ~ factor * 2**(n-1) seconds between retries, plus jitter
+BACKOFF_JITTER = 1.0
+BACKOFF_MAX = 30.0
+RETRY_STATUSES = tuple(range(500, 600))
+EXIT_CODE_PGT_FAILURE = 78  # distinct job exit status for PGT token failures
+
+
+class PgtTokenError(Exception):
+    """Raised when the MAAP PGT token cannot be obtained."""
 
 
-def get_maap_pgt_token(username, maap_api_host=os.environ.get("MAAP_API_HOST")):
+def _session():
+    """Session retrying connect errors, timeouts and 5xx with exponential backoff and jitter."""
+    retry = Retry(
+        total=MAX_RETRIES,
+        backoff_factor=BACKOFF_FACTOR,
+        backoff_jitter=BACKOFF_JITTER,
+        backoff_max=BACKOFF_MAX,
+        status_forcelist=RETRY_STATUSES,
+        raise_on_status=False,  # hand back the last response so we can raise a clear error
+    )
+    session = requests.Session()
+    adapter = HTTPAdapter(max_retries=retry)
+    session.mount("https://", adapter)
+    session.mount("http://", adapter)
+    return session
+
+
+def get_maap_pgt_token(username, maap_api_host=None):
+    maap_api_host = maap_api_host or os.environ.get("MAAP_API_HOST")
     url = f"https://{maap_api_host}/api/members/{username}"
     headers = {"dps-token": os.environ.get("DPS_MACHINE_TOKEN")}
-    response = requests.get(url, headers=headers)
-    maap_pgt_token = response.json()["session_key"]
-    return maap_pgt_token
+
+    try:
+        with _session() as session:
+            response = session.get(url, headers=headers, timeout=TIMEOUT)
+    except requests.RequestException as e:
+        raise PgtTokenError(
+            f"Could not get MAAP_PGT for user '{username}' from {url} "
+            f"after {MAX_RETRIES + 1} attempts; last error: {type(e).__name__}: {e}"
+        ) from e
+
+    if response.status_code >= 500:
+        raise PgtTokenError(
+            f"Could not get MAAP_PGT for user '{username}' from {url} "
+            f"after {MAX_RETRIES + 1} attempts; last error: HTTP {response.status_code}"
+        )
+
+    try:
+        response.raise_for_status()
+    except requests.HTTPError as e:
+        # 4xx is permanent (e.g. bad token, unknown user) and was not retried
+        raise PgtTokenError(
+            f"Request for user '{username}' to {url} failed: HTTP {response.status_code}"
+        ) from e
+
+    try:
+        token = response.json()["session_key"]
+    except (ValueError, KeyError, TypeError) as e:
+        raise PgtTokenError(
+            f"Response from {url} for user '{username}' has no usable 'session_key'"
+        ) from e
+    if not isinstance(token, str) or not token.strip():
+        raise PgtTokenError(
+            f"Response from {url} for user '{username}' has an empty 'session_key'"
+        )
+    return token
+
+
+def report_failure(error, job_dir="."):
+    """Surface the failure to HySDS.
+
+    stderr lands in _stderr.txt; _alt_error.txt overrides the job's `error`
+    field (job_worker reads it from the job dir). Keep it a short single line.
+    """
+    message = " ".join(str(error).split())
+    print(f"Unable to get user MAAP_PGT token: {message}", file=sys.stderr)
+    with open(os.path.join(job_dir, "_alt_error.txt"), "w") as f:
+        f.write(f"MAAP_PGT token failure: {message}\n")
 
 
 if __name__ == '__main__':
     with open("_job.json") as fr:
         job = json.loads(fr.read())
-        username = job.get("username")
-        if username:
-            try:
-                maap_pgt_token = get_maap_pgt_token(username)
-                print(maap_pgt_token)
-            except Exception as e:
-                print(f"Unable to get user MAAP_PGT token: {e}")
-                raise e
+    username = job.get("username")
+    try:
+        if not username:
+            # an empty MAAP_PGT would only surface later as a 401 in the user's code
+            raise PgtTokenError("no 'username' in _job.json, cannot request a MAAP_PGT")
+        print(get_maap_pgt_token(username))
+    except PgtTokenError as e:
+        # stdout is captured into MAAP_PGT by dps_wrapper.sh, so report on stderr
+        report_failure(e)
+        sys.exit(EXIT_CODE_PGT_FAILURE)
diff --git a/stage_in.py b/stage_in.py
index 509adc9..b146c0f 100644
--- a/stage_in.py
+++ b/stage_in.py
@@ -7,11 +7,19 @@
 from datetime import datetime
 from maap.maap import MAAP
 
-maap = MAAP()
+_maap = None
+
+
+def _get_maap():
+    """Create the MAAP client on first use so jobs without file downloads make no API call."""
+    global _maap
+    if _maap is None:
+        _maap = MAAP()
+    return _maap
 
 
 def _download_asset(href, destination_path):
-    """Route asset download: MAAP S3 buckets use boto3, everything else uses maap.downloadGranule."""
+    """Route asset download: MAAP S3 buckets use boto3, everything else uses maap.download_granule."""
     if href.startswith("s3://maap-"):
         parts = href[5:].split("/", 1)
         bucket, key = parts[0], parts[1]
@@ -19,7 +27,7 @@ def _download_asset(href, destination_path):
         dest = os.path.join(destination_path, filename)
         boto3.client("s3").download_file(bucket, key, dest)
         return dest
-    return maap.downloadGranule(
+    return _get_maap().download_granule(
         online_access_url=href,
         destination_path=destination_path,
         overwrite=True,
diff --git a/tests/test_create_inputs.py b/tests/test_create_inputs.py
new file mode 100644
index 0000000..f4c686f
--- /dev/null
+++ b/tests/test_create_inputs.py
@@ -0,0 +1,102 @@
+import pytest
+import sys
+from pathlib import Path
+from unittest.mock import MagicMock, Mock, patch
+
+# Add parent directory to path to import create_inputs
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+from create_inputs import map_inputs_from_context
+
+
+def _cwl_input(name, type_):
+    """Build a minimal stand-in for a cwl_utils CommandInputParameter."""
+    return Mock(id=f"file:///process.cwl#main/{name}", type_=type_)
+
+
+class TestOptionalFileDirectoryInputs:
+    """Test handling of empty File/Directory inputs in map_inputs_from_context()."""
+
+    @pytest.mark.parametrize("type_", [["null", "File"], "File?"])
+    @pytest.mark.parametrize("value", [None, "", "None"])
+    def test_empty_optional_file_is_omitted(self, type_, value):
+        cwl_inputs = [_cwl_input("opt_file", type_)]
+        job_inputs = {} if value is None else {"opt_file": value}
+
+        assert map_inputs_from_context(job_inputs, cwl_inputs) == {}
+
+    @pytest.mark.parametrize("type_", [["null", "Directory"], "Directory?"])
+    @pytest.mark.parametrize("value", [None, "", "None"])
+    @patch('create_inputs.stage_in.download_stac_assets_from_json')
+    @patch('create_inputs.load_json_from_uri')
+    def test_empty_optional_directory_is_omitted_without_staging(self, mock_load, mock_download, type_, value):
+        cwl_inputs = [_cwl_input("opt_dir", type_)]
+        job_inputs = {} if value is None else {"opt_dir": value}
+
+        assert map_inputs_from_context(job_inputs, cwl_inputs) == {}
+        mock_load.assert_not_called()
+        mock_download.assert_not_called()
+
+    @pytest.mark.parametrize("type_", ["File", "Directory"])
+    @pytest.mark.parametrize("value", [None, ""])
+    def test_empty_required_file_or_directory_raises(self, type_, value):
+        cwl_inputs = [_cwl_input("req_input", type_)]
+        job_inputs = {} if value is None else {"req_input": value}
+
+        with pytest.raises(ValueError, match="req_input"):
+            map_inputs_from_context(job_inputs, cwl_inputs)
+
+    def test_provided_optional_file_is_included(self):
+        cwl_inputs = [_cwl_input("opt_file", ["null", "File"])]
+        job_inputs = {"opt_file": "s3://bucket/path/data.tif"}
+
+        result = map_inputs_from_context(job_inputs, cwl_inputs)
+
+        assert result["opt_file"]["class"] == "File"
+        assert result["opt_file"]["path"] == str(Path("data.tif").resolve())
+
+    @patch('create_inputs.stage_in.download_stac_assets_from_json')
+    @patch('create_inputs.load_json_from_uri', return_value={"type": "Feature"})
+    def test_provided_optional_directory_is_staged(self, mock_load, mock_download):
+        cwl_inputs = [_cwl_input("opt_dir", ["null", "Directory"])]
+        job_inputs = {"opt_dir": "https://example.com/item.json"}
+
+        result = map_inputs_from_context(job_inputs, cwl_inputs)
+
+        mock_load.assert_called_once_with("https://example.com/item.json")
+        mock_download.assert_called_once()
+        assert result["opt_dir"] == {"class": "Directory", "path": str(Path("opt_dir").resolve())}
+
+    def test_empty_optional_file_does_not_affect_other_inputs(self):
+        cwl_inputs = [
+            _cwl_input("opt_file", ["null", "File"]),
+            _cwl_input("req_str", "string"),
+            _cwl_input("opt_str", ["null", "string"]),
+        ]
+        job_inputs = {"req_str": "x", "opt_str": ""}
+
+        assert map_inputs_from_context(job_inputs, cwl_inputs) == {"req_str": "x"}
+
+
+class TestLoadJsonFromUriRetry:
+    @patch('create_inputs.time.sleep')
+    @patch('create_inputs.requests.get')
+    def test_retries_transient_503_then_succeeds(self, mock_get, _sleep):
+        import requests
+        import create_inputs
+        bad = MagicMock()
+        bad.raise_for_status.side_effect = requests.exceptions.HTTPError("503")
+        good = MagicMock()
+        good.json.return_value = {"type": "Feature"}
+        mock_get.side_effect = [bad, bad, good]
+        assert create_inputs.load_json_from_uri("https://example.com/item") == {"type": "Feature"}
+        assert mock_get.call_count == 3
+
+    @patch('create_inputs.time.sleep')
+    @patch('create_inputs.requests.get')
+    def test_returns_none_after_exhausting_retries(self, mock_get, _sleep):
+        import requests
+        import create_inputs
+        mock_get.side_effect = requests.exceptions.ConnectionError("down")
+        assert create_inputs.load_json_from_uri("https://example.com/item") is None
+        assert mock_get.call_count == create_inputs.HTTP_RETRIES
diff --git a/tests/test_get_maap_pgt_token.py b/tests/test_get_maap_pgt_token.py
new file mode 100644
index 0000000..52aab74
--- /dev/null
+++ b/tests/test_get_maap_pgt_token.py
@@ -0,0 +1,189 @@
+import json
+import subprocess
+import sys
+import threading
+from http.server import BaseHTTPRequestHandler, HTTPServer
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+import pytest
+import requests
+
+sys.path.insert(0, str(Path(__file__).parent.parent))
+
+import get_maap_pgt_token as mod
+from get_maap_pgt_token import get_maap_pgt_token, PgtTokenError
+
+
+def _resp(status=200, body=None):
+    r = MagicMock()
+    r.status_code = status
+    r.json.return_value = body if body is not None else {}
+    if status >= 400:
+        r.raise_for_status.side_effect = requests.HTTPError(f"{status}")
+    return r
+
+
+def _mock_session(get):
+    """Patch mod._session to return a context-manager session whose .get is `get`."""
+    session = MagicMock()
+    session.__enter__.return_value = session
+    session.get = get
+    return patch.object(mod, "_session", return_value=session)
+
+
+@pytest.fixture(autouse=True)
+def env(monkeypatch):
+    monkeypatch.setenv("MAAP_API_HOST", "api.example.org")
+    monkeypatch.setenv("DPS_MACHINE_TOKEN", "tok")
+
+
+@pytest.fixture
+def fast_backoff(monkeypatch):
+    monkeypatch.setattr(mod, "BACKOFF_FACTOR", 0.01)
+    monkeypatch.setattr(mod, "BACKOFF_JITTER", 0.01)
+
+
+@pytest.fixture
+def flaky_server():
+    """Local HTTP server; set .statuses to the sequence of codes to return, then 200."""
+    class Handler(BaseHTTPRequestHandler):
+        def do_GET(self):
+            self.server.hits += 1
+            code = self.server.statuses.pop(0) if self.server.statuses else 200
+            body = json.dumps({"session_key": "abc"} if code == 200 else {}).encode()
+            self.send_response(code)
+            self.send_header("Content-Length", str(len(body)))
+            self.end_headers()
+            self.wfile.write(body)
+
+        def log_message(self, *a):
+            pass
+
+    server = HTTPServer(("127.0.0.1", 0), Handler)
+    server.hits, server.statuses = 0, []
+    threading.Thread(target=server.serve_forever, daemon=True).start()
+    server.url = f"http://127.0.0.1:{server.server_port}/"
+    yield server
+    server.shutdown()
+
+
+# --- retry behaviour, against a real server through the real session ---
+
+def test_session_retries_5xx_then_succeeds(flaky_server, fast_backoff):
+    flaky_server.statuses = [503, 502]
+    with mod._session() as s:
+        r = s.get(flaky_server.url, timeout=mod.TIMEOUT)
+    assert r.status_code == 200
+    assert flaky_server.hits == 3
+
+
+def test_session_gives_up_after_max_retries_returning_last_response(flaky_server, fast_backoff):
+    flaky_server.statuses = [500] * 20
+    with mod._session() as s:
+        r = s.get(flaky_server.url, timeout=mod.TIMEOUT)
+    assert r.status_code == 500
+    assert flaky_server.hits == mod.MAX_RETRIES + 1
+
+
+def test_session_does_not_retry_4xx(flaky_server, fast_backoff):
+    flaky_server.statuses = [401, 200]
+    with mod._session() as s:
+        r = s.get(flaky_server.url, timeout=mod.TIMEOUT)
+    assert r.status_code == 401
+    assert flaky_server.hits == 1
+
+
+def test_session_retries_connection_errors(fast_backoff):
+    with mod._session() as s:
+        with pytest.raises(requests.ConnectionError):
+            s.get("http://127.0.0.1:9/", timeout=mod.TIMEOUT)
+
+
+# --- get_maap_pgt_token logic ---
+
+def test_success_passes_timeout_and_header():
+    get = MagicMock(return_value=_resp(200, {"session_key": "abc"}))
+    with _mock_session(get):
+        assert get_maap_pgt_token("alice") == "abc"
+    get.assert_called_once_with(
+        "https://api.example.org/api/members/alice",
+        headers={"dps-token": "tok"},
+        timeout=mod.TIMEOUT,
+    )
+
+
+def test_exhausted_5xx_gives_clear_error():
+    with _mock_session(MagicMock(return_value=_resp(503))):
+        with pytest.raises(PgtTokenError, match=r"after 6 attempts.*HTTP 503"):
+            get_maap_pgt_token("alice")
+
+
+def test_exhausted_connection_errors_give_clear_error():
+    with _mock_session(MagicMock(side_effect=requests.ConnectionError("boom"))):
+        with pytest.raises(PgtTokenError, match=r"after 6 attempts.*ConnectionError: boom"):
+            get_maap_pgt_token("alice")
+
+
+def test_4xx_gives_clear_error():
+    with _mock_session(MagicMock(return_value=_resp(401))):
+        with pytest.raises(PgtTokenError, match="HTTP 401"):
+            get_maap_pgt_token("alice")
+
+
+def test_missing_session_key_gives_clear_error():
+    with _mock_session(MagicMock(return_value=_resp(200, {"other": 1}))):
+        with pytest.raises(PgtTokenError, match="session_key"):
+            get_maap_pgt_token("alice")
+
+
+def test_non_json_body_gives_clear_error():
+    r = _resp(200)
+    r.json.side_effect = ValueError("not json")
+    with _mock_session(MagicMock(return_value=r)):
+        with pytest.raises(PgtTokenError, match="session_key"):
+            get_maap_pgt_token("alice")
+
+
+@pytest.mark.parametrize("key", ["", "   ", None, 123])
+def test_empty_or_non_string_session_key_gives_clear_error(key):
+    with _mock_session(MagicMock(return_value=_resp(200, {"session_key": key}))):
+        with pytest.raises(PgtTokenError, match="empty 'session_key'"):
+            get_maap_pgt_token("alice")
+
+
+# --- failure reporting to HySDS ---
+
+def test_report_failure_writes_stderr_and_alt_error(tmp_path, capsys):
+    mod.report_failure(PgtTokenError("HTTP 503\nafter 6 attempts"), job_dir=str(tmp_path))
+    assert "HTTP 503 after 6 attempts" in capsys.readouterr().err
+    assert (tmp_path / "_alt_error.txt").read_text() == "MAAP_PGT token failure: HTTP 503 after 6 attempts\n"
+
+
+def test_main_exits_with_custom_code_and_keeps_stdout_clean(tmp_path):
+    (tmp_path / "_job.json").write_text(json.dumps({"username": "alice"}))
+    # nothing listens on localhost:9; sleep is stubbed so retries are exhausted quickly
+    code = (
+        "import runpy, time; time.sleep = lambda s: None; "
+        f"runpy.run_path({mod.__file__!r}, run_name='__main__')"
+    )
+    r = subprocess.run(
+        [sys.executable, "-c", code], cwd=tmp_path, capture_output=True, text=True,
+        env={"MAAP_API_HOST": "localhost:9", "DPS_MACHINE_TOKEN": "t", "PATH": "",
+             "PYTHONPATH": str(Path(mod.__file__).parent)},
+    )
+    assert r.returncode == mod.EXIT_CODE_PGT_FAILURE, r.stderr
+    assert r.stdout == ""
+    assert "Unable to get user MAAP_PGT token" in r.stderr
+    assert (tmp_path / "_alt_error.txt").exists()
+
+
+def test_main_without_username_fails_instead_of_printing_empty_token(tmp_path):
+    (tmp_path / "_job.json").write_text(json.dumps({}))
+    r = subprocess.run(
+        [sys.executable, mod.__file__], cwd=tmp_path, capture_output=True, text=True,
+        env={"PATH": ""},
+    )
+    assert r.returncode == mod.EXIT_CODE_PGT_FAILURE, r.stderr
+    assert r.stdout == ""
+    assert "username" in (tmp_path / "_alt_error.txt").read_text()
diff --git a/tests/test_stage_in.py b/tests/test_stage_in.py
index 41cb2d1..b71ba28 100644
--- a/tests/test_stage_in.py
+++ b/tests/test_stage_in.py
@@ -205,8 +205,8 @@ def test_metadata_missing_keys_use_defaults(self, tmp_path):
 class TestSingleItemDownload:
     """Test _download_single_item() behavior."""
     
-    @patch('stage_in.maap.downloadGranule')
-    def test_subdirectory_created(self, mock_download, tmp_path):
+    @patch('stage_in._get_maap')
+    def test_subdirectory_created(self, mock_get_maap, tmp_path):
         """A subdirectory <output_dir>/<item.id>/ should be created."""
         item_json = {
             "type": "Feature",
@@ -226,10 +226,10 @@ def test_subdirectory_created(self, mock_download, tmp_path):
         assert expected_dir.exists()
         assert expected_dir.is_dir()
     
-    @patch('stage_in.maap.downloadGranule')
-    def test_download_granule_called_per_asset(self, mock_download, tmp_path):
-        """maap.downloadGranule should be called once per asset with the correct URL."""
-        mock_download.side_effect = [
+    @patch('stage_in._get_maap')
+    def test_download_granule_called_per_asset(self, mock_get_maap, tmp_path):
+        """maap.download_granule should be called once per asset with the correct URL."""
+        mock_get_maap.return_value.download_granule.side_effect = [
             str(tmp_path / "test-item-123" / "asset1.tif"),
             str(tmp_path / "test-item-123" / "asset2.json")
         ]
@@ -251,15 +251,15 @@ def test_download_granule_called_per_asset(self, mock_download, tmp_path):
         parent_container = Mock()
         _download_single_item(item_json, str(tmp_path), parent_container)
         
-        assert mock_download.call_count == 2
-        calls = mock_download.call_args_list
+        assert mock_get_maap.return_value.download_granule.call_count == 2
+        calls = mock_get_maap.return_value.download_granule.call_args_list
         assert calls[0][1]['online_access_url'] == "https://example.com/data.tif"
         assert calls[1][1]['online_access_url'] == "https://example.com/meta.json"
     
-    @patch('stage_in.maap.downloadGranule')
-    def test_returned_item_added_to_parent(self, mock_download, tmp_path):
+    @patch('stage_in._get_maap')
+    def test_returned_item_added_to_parent(self, mock_get_maap, tmp_path):
         """The returned item should be added to the parent container."""
-        mock_download.return_value = str(tmp_path / "test-item-123" / "asset.tif")
+        mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-123" / "asset.tif")
 
         item_json = {
             "type": "Feature",
@@ -282,10 +282,10 @@ def test_returned_item_added_to_parent(self, mock_download, tmp_path):
         assert added_item.id == "test-item-123"
         assert isinstance(result, pystac.Item)
     
-    @patch('stage_in.maap.downloadGranule')
-    def test_asset_hrefs_are_relative(self, mock_download, tmp_path):
+    @patch('stage_in._get_maap')
+    def test_asset_hrefs_are_relative(self, mock_get_maap, tmp_path):
         """All asset hrefs on the returned item should be relative paths."""
-        mock_download.return_value = str(tmp_path / "test-item-123" / "asset.tif")
+        mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-123" / "asset.tif")
 
         item_json = {
             "type": "Feature",
@@ -307,8 +307,8 @@ def test_asset_hrefs_are_relative(self, mock_download, tmp_path):
             assert not asset.href.startswith("http")
             assert not os.path.isabs(asset.href)
     
-    @patch('stage_in.maap.downloadGranule')
-    def test_one_asset_fails_rest_still_download(self, mock_download, tmp_path):
+    @patch('stage_in._get_maap')
+    def test_one_asset_fails_rest_still_download(self, mock_get_maap, tmp_path):
         """If one asset fails, the rest should still download (no crash)."""
         def side_effect(*args, **kwargs):
             url = kwargs['online_access_url']
@@ -316,7 +316,7 @@ def side_effect(*args, **kwargs):
                 raise Exception("Download failed")
             return str(tmp_path / "test-item-123" / "success.tif")
 
-        mock_download.side_effect = side_effect
+        mock_get_maap.return_value.download_granule.side_effect = side_effect
 
         item_json = {
             "type": "Feature",
@@ -336,15 +336,15 @@ def side_effect(*args, **kwargs):
         parent_container = Mock()
         result = _download_single_item(item_json, str(tmp_path), parent_container)
         
-        assert mock_download.call_count == 3
+        assert mock_get_maap.return_value.download_granule.call_count == 3
         assert len(result.assets) == 2
         assert "good1" in result.assets
         assert "good2" in result.assets
         assert "bad" not in result.assets
     
-    @patch('stage_in.maap.downloadGranule')
-    def test_zero_assets_no_download_calls(self, mock_download, tmp_path):
-        """Zero assets should result in downloadGranule never being called."""
+    @patch('stage_in._get_maap')
+    def test_zero_assets_no_download_calls(self, mock_get_maap, tmp_path):
+        """Zero assets should result in download_granule never being called."""
         item_json = {
             "type": "Feature",
             "stac_version": "1.0.0",
@@ -359,16 +359,16 @@ def test_zero_assets_no_download_calls(self, mock_download, tmp_path):
         parent_container = Mock()
         _download_single_item(item_json, str(tmp_path), parent_container)
         
-        mock_download.assert_not_called()
+        mock_get_maap.return_value.download_granule.assert_not_called()
 
 
 class TestStacMetadataPreservation:
     """Test preservation of STAC metadata during item cloning and local staging."""
 
-    @patch('stage_in.maap.downloadGranule')
-    def test_extra_fields_preserved(self, mock_download, tmp_path):
+    @patch('stage_in._get_maap')
+    def test_extra_fields_preserved(self, mock_get_maap, tmp_path):
         """Extra fields (non-standard STAC properties) should be preserved via clone() and explicit update."""
-        mock_download.return_value = str(tmp_path / "test-item-meta" / "asset.tif")
+        mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-meta" / "asset.tif")
 
         item_json = {
             "type": "Feature",
@@ -392,10 +392,10 @@ def test_extra_fields_preserved(self, mock_download, tmp_path):
         assert result.extra_fields.get("custom_field") == "custom_value"
         assert result.extra_fields.get("another_custom") == {"nested": "object"}
 
-    @patch('stage_in.maap.downloadGranule')
-    def test_extra_fields_survive_serialization(self, mock_download, tmp_path):
+    @patch('stage_in._get_maap')
+    def test_extra_fields_survive_serialization(self, mock_get_maap, tmp_path):
         """Extra fields should survive to_dict() serialization for JSON output."""
-        mock_download.return_value = str(tmp_path / "test-item-meta" / "asset.tif")
+        mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-meta" / "asset.tif")
 
         item_json = {
             "type": "Feature",
@@ -420,10 +420,10 @@ def test_extra_fields_survive_serialization(self, mock_download, tmp_path):
         assert result_dict.get("custom_field") == "custom_value"
         assert result_dict.get("license") == "CC-BY-4.0"
 
-    @patch('stage_in.maap.downloadGranule')
-    def test_stac_extensions_preserved(self, mock_download, tmp_path):
+    @patch('stage_in._get_maap')
+    def test_stac_extensions_preserved(self, mock_get_maap, tmp_path):
         """stac_extensions list should be preserved during clone."""
-        mock_download.return_value = str(tmp_path / "test-item-meta" / "asset.tif")
+        mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-meta" / "asset.tif")
 
         item_json = {
             "type": "Feature",
@@ -445,15 +445,16 @@ def test_stac_extensions_preserved(self, mock_download, tmp_path):
         parent_container = Mock()
         result = _download_single_item(item_json, str(tmp_path), parent_container)
 
-        assert result.stac_extensions == [
-            "https://stac-extensions.github.io/eo/v1.0.0/schema.json",
-            "https://stac-extensions.github.io/sar/v0.9.0/schema.json"
-        ]
+        # pystac may migrate extension schemas to newer versions on read,
+        # so check that each extension is preserved without pinning its version.
+        assert len(result.stac_extensions) == 2
+        assert any("/eo/" in ext for ext in result.stac_extensions)
+        assert any("/sar/" in ext for ext in result.stac_extensions)
 
-    @patch('stage_in.maap.downloadGranule')
-    def test_stac_extensions_empty_list_preserved(self, mock_download, tmp_path):
+    @patch('stage_in._get_maap')
+    def test_stac_extensions_empty_list_preserved(self, mock_get_maap, tmp_path):
         """Empty stac_extensions list should be preserved."""
-        mock_download.return_value = str(tmp_path / "test-item-meta" / "asset.tif")
+        mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-meta" / "asset.tif")
 
         item_json = {
             "type": "Feature",
@@ -474,10 +475,10 @@ def test_stac_extensions_empty_list_preserved(self, mock_download, tmp_path):
 
         assert result.stac_extensions == []
 
-    @patch('stage_in.maap.downloadGranule')
-    def test_original_item_dict_not_mutated(self, mock_download, tmp_path):
+    @patch('stage_in._get_maap')
+    def test_original_item_dict_not_mutated(self, mock_get_maap, tmp_path):
         """The input item_json dict should not be mutated by clone()."""
-        mock_download.return_value = str(tmp_path / "test-item-meta" / "asset.tif")
+        mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-meta" / "asset.tif")
 
         original_href = "https://example.com/data.tif"
         item_json = {
@@ -499,10 +500,10 @@ def test_original_item_dict_not_mutated(self, mock_download, tmp_path):
         # Input dict should retain original remote href
         assert item_json["assets"]["data"]["href"] == original_href
 
-    @patch('stage_in.maap.downloadGranule')
-    def test_links_cleared_and_self_href_reset(self, mock_download, tmp_path):
+    @patch('stage_in._get_maap')
+    def test_links_cleared_and_self_href_reset(self, mock_get_maap, tmp_path):
         """Remote links should be cleared and replaced with local self href."""
-        mock_download.return_value = str(tmp_path / "test-item-meta" / "asset.tif")
+        mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-meta" / "asset.tif")
 
         item_json = {
             "type": "Feature",
@@ -528,10 +529,10 @@ def test_links_cleared_and_self_href_reset(self, mock_download, tmp_path):
         assert result.links[0].rel == "self"
         assert "example.com" not in result.links[0].target
 
-    @patch('stage_in.maap.downloadGranule')
-    def test_self_href_set_to_local_path(self, mock_download, tmp_path):
-        """Self href should be set to the resolved absolute path of <output_dir>/<item.id>."""
-        mock_download.return_value = str(tmp_path / "test-item-meta" / "asset.tif")
+    @patch('stage_in._get_maap')
+    def test_self_href_set_to_local_path(self, mock_get_maap, tmp_path):
+        """Self href should be the resolved absolute path of <output_dir>/<item.id>/<item.id>.json."""
+        mock_get_maap.return_value.download_granule.return_value = str(tmp_path / "test-item-meta" / "asset.tif")
 
         item_json = {
             "type": "Feature",
@@ -549,7 +550,7 @@ def test_self_href_set_to_local_path(self, mock_download, tmp_path):
         parent_container = Mock()
         result = _download_single_item(item_json, str(tmp_path), parent_container)
 
-        expected_self_href = str((tmp_path / "test-item-meta").resolve())
+        expected_self_href = str((tmp_path / "test-item-meta" / "test-item-meta.json").resolve())
         actual_self_href = result.get_self_href()
         assert actual_self_href == expected_self_href
 
@@ -557,9 +558,9 @@ def test_self_href_set_to_local_path(self, mock_download, tmp_path):
 class TestSingleItemCatalogAssembly:
     """Test download_stac_item_assets_from_json() catalog assembly."""
 
-    @patch('stage_in.maap.downloadGranule')
+    @patch('stage_in._get_maap')
     @patch('pystac.Catalog.save')
-    def test_output_directory_created(self, mock_save, mock_download, tmp_path):
+    def test_output_directory_created(self, mock_save, mock_get_maap, tmp_path):
         """Output directory should be created."""
         output_dir = tmp_path / "test_output"
 
@@ -578,9 +579,9 @@ def test_output_directory_created(self, mock_save, mock_download, tmp_path):
         
         assert output_dir.exists()
     
-    @patch('stage_in.maap.downloadGranule')
+    @patch('stage_in._get_maap')
     @patch('pystac.Catalog.save')
-    def test_catalog_json_exists(self, mock_save, mock_download, tmp_path):
+    def test_catalog_json_exists(self, mock_save, mock_get_maap, tmp_path):
         """catalog.json should exist after the call."""
         output_dir = tmp_path / "test_output"
 
@@ -599,12 +600,12 @@ def test_catalog_json_exists(self, mock_save, mock_download, tmp_path):
         
         mock_save.assert_called_once()
     
-    @patch('stage_in.maap.downloadGranule')
+    @patch('stage_in._get_maap')
     @patch('pystac.Catalog.save')
-    def test_catalog_contains_one_item(self, mock_save, mock_download, tmp_path):
+    def test_catalog_contains_one_item(self, mock_save, mock_get_maap, tmp_path):
         """The catalog should contain exactly the one item from input."""
         output_dir = tmp_path / "test_output"
-        mock_download.return_value = str(output_dir / "test-item" / "asset.tif")
+        mock_get_maap.return_value.download_granule.return_value = str(output_dir / "test-item" / "asset.tif")
 
         item_json = {
             "type": "Feature",
@@ -831,10 +832,10 @@ def test_feature_collection_json_file_parsed_and_dispatched(self, mock_dispatch,
 
 
 class TestAssetRouting:
-    """Test _download_asset() routes to boto3 vs maap.downloadGranule based on URL."""
+    """Test _download_asset() routes to boto3 vs maap.download_granule based on URL."""
 
     @patch('stage_in.boto3')
-    @patch('stage_in.maap')
+    @patch('stage_in._get_maap')
     def test_maap_s3_url_uses_boto3(self, mock_maap, mock_boto3, tmp_path):
         href = "s3://maap-data-store/path/to/file.tif"
         mock_s3 = MagicMock()
@@ -845,30 +846,69 @@ def test_maap_s3_url_uses_boto3(self, mock_maap, mock_boto3, tmp_path):
 
         mock_boto3.client.assert_called_once_with("s3")
         mock_s3.download_file.assert_called_once_with("maap-data-store", "path/to/file.tif", str(tmp_path / "file.tif"))
-        mock_maap.downloadGranule.assert_not_called()
+        mock_maap.return_value.download_granule.assert_not_called()
         assert dest == str(tmp_path / "file.tif")
 
     @patch('stage_in.boto3')
-    @patch('stage_in.maap')
+    @patch('stage_in._get_maap')
     def test_daac_url_uses_maap_download_granule(self, mock_maap, mock_boto3):
         href = "https://daac.gsfc.nasa.gov/data/file.tif"
-        mock_maap.downloadGranule.return_value = "/local/file.tif"
+        mock_maap.return_value.download_granule.return_value = "/local/file.tif"
 
         dest = _download_asset(href, "/local")
 
-        mock_maap.downloadGranule.assert_called_once_with(
+        mock_maap.return_value.download_granule.assert_called_once_with(
             online_access_url=href, destination_path="/local", overwrite=True
         )
         mock_boto3.client.assert_not_called()
         assert dest == "/local/file.tif"
 
     @patch('stage_in.boto3')
-    @patch('stage_in.maap')
+    @patch('stage_in._get_maap')
     def test_non_maap_s3_url_falls_through_to_maap(self, mock_maap, mock_boto3):
         href = "s3://some-other-bucket/data/file.tif"
-        mock_maap.downloadGranule.return_value = "/local/file.tif"
+        mock_maap.return_value.download_granule.return_value = "/local/file.tif"
 
         _download_asset(href, "/local")
 
-        mock_maap.downloadGranule.assert_called_once()
+        mock_maap.return_value.download_granule.assert_called_once()
         mock_boto3.client.assert_not_called()
+
+
+class TestLazyClient:
+    """The MAAP client must not be constructed at import or for non-MAAP downloads."""
+
+    @patch('stage_in.boto3')
+    @patch('stage_in.MAAP')
+    def test_no_client_for_maap_s3(self, mock_cls, mock_boto3, tmp_path):
+        import stage_in
+        stage_in._maap = None
+        mock_boto3.client.return_value.download_file.side_effect = lambda b, k, d: open(d, 'w').close()
+        _download_asset("s3://maap-data-store/a/file.tif", str(tmp_path))
+        mock_cls.assert_not_called()
+
+    @patch('stage_in.MAAP')
+    def test_client_created_once_on_demand(self, mock_cls):
+        import stage_in
+        stage_in._maap = None
+        _download_asset("https://example.com/f.tif", "/tmp")
+        _download_asset("https://example.com/g.tif", "/tmp")
+        mock_cls.assert_called_once()
+        stage_in._maap = None
+
+
+class TestMaapPyContract:
+    """Guard against maap-py API drift: stage_in must call a method the installed SDK really has."""
+
+    def test_download_granule_signature(self):
+        import inspect
+        from maap.maap import MAAP
+        params = inspect.signature(MAAP.download_granule).parameters
+        assert {"online_access_url", "destination_path", "overwrite"} <= set(params)
+
+    @patch('stage_in._get_maap')
+    def test_download_asset_uses_download_granule(self, mock_get_maap):
+        import stage_in
+        stage_in._download_asset("https://example.com/a.tif", "/tmp/dest")
+        mock_get_maap.return_value.download_granule.assert_called_once_with(
+            online_access_url="https://example.com/a.tif", destination_path="/tmp/dest", overwrite=True)