diff --git a/spark_executor/core/yarn_client.py b/spark_executor/core/yarn_client.py
index fc32c60..7073b90 100644
--- a/spark_executor/core/yarn_client.py
+++ b/spark_executor/core/yarn_client.py
@@ -8,9 +8,15 @@ so the runtime image does not need a Hadoop client installation — the
 `httpx` library already in pyproject.toml is enough.
 
 Endpoints used (YARN 2.6+):
-  GET  /ws/v1/cluster/apps/{appid}             -> app status + state
+  GET  /ws/v1/cluster/apps/{appid}                   -> app status + state; amContainerLogs
   GET  /ws/v1/cluster/apps/{appid}/aggregated-logs  -> aggregated container logs
-  PUT  /ws/v1/cluster/apps/{appid}/state       -> kill an app (body: {"state":"KILLED"})
+  PUT  /ws/v1/cluster/apps/{appid}/state             -> kill an app (body: {"state":"KILLED"})
+
+When aggregated logs are unavailable (HTTP 404/501, e.g. log aggregation
+disabled), fall back to the amContainerLogs field reported by the app
+endpoint and fetch the AM (driver) container's /stdout directly from the
+NodeManager. This returns driver logs only; executor container logs still
+require yarn.log-aggregation-enable=true (the primary /aggregated-logs path).
 
 The ResourceManager URL is passed in per call (snapshotted on the Job at
 confirm_submit_job time) and falls back to the YARN_RESOURCE_MANAGER_URL env
@@ -157,55 +163,32 @@ def _logs_unavailable_error(application_id: str) -> YarnError:
     )
 
 
-def _fetch_logs_via_nodemanager(application_id: str, config: YarnClientConfig) -> str:
+def _fetch_logs_via_am_container(application_id: str, config: YarnClientConfig) -> str:
     """
-    Hadoop 2.x fallback: walk app attempts -> containers -> NodeManager
-    container logs and concatenate the results.
+    Fetch ApplicationMaster (driver) container stdout as a fallback when the
+    aggregated-logs endpoint is unavailable.
+
+    Returns AM (driver) container logs only. Executor container logs require
+    yarn.log-aggregation-enable=true, which is covered by the primary
+    /aggregated-logs path.
     """
     base = _base_url(config.yarn_rm_url)
     app_url = f"{base}/ws/v1/cluster/apps/{application_id}"
 
-    # Latest (or all) app attempts.
-    attempts_url = f"{app_url}/appattempts"
-    resp = _request("GET", attempts_url, timeout=30.0, verify=config.verify_for_httpx(), auth=config.auth_for_httpx())
+    resp = _request("GET", app_url, verify=config.verify_for_httpx(), auth=config.auth_for_httpx())
     if resp.status_code >= 400:
         raise _logs_unavailable_error(application_id)
-    attempts = resp.json().get("appAttempts", {}).get("appAttempt", [])
-    if not attempts:
+
+    am_container_logs = resp.json().get("app", {}).get("amContainerLogs")
+    if not am_container_logs:
         raise _logs_unavailable_error(application_id)
 
-    # Collect containers across attempts.
-    containers: list[dict] = []
-    for attempt in attempts:
-        attempt_id = attempt.get("id")
-        if not attempt_id:
-            continue
-        containers_url = f"{app_url}/appattempts/{attempt_id}/containers"
-        resp = _request("GET", containers_url, timeout=30.0, verify=config.verify_for_httpx(), auth=config.auth_for_httpx())
-        if resp.status_code >= 400:
-            raise _logs_unavailable_error(application_id)
-        containers.extend(resp.json().get("containers", {}).get("container", []))
-
-    if not containers:
+    log_url = f"{am_container_logs.rstrip('/')}/stdout"
+    resp = _request("GET", log_url, timeout=60.0, verify=config.verify_for_httpx(), auth=config.auth_for_httpx())
+    if resp.status_code >= 400:
+        logger.error(f"YARN GET {log_url} -> {resp.status_code}: {resp.text[:500]}")
         raise _logs_unavailable_error(application_id)
-
-    parts = []
-    for container in containers:
-        container_id = container.get("id")
-        user = container.get("user")
-        node_http_address = container.get("nodeHttpAddress")
-        if not all((container_id, user, node_http_address)):
-            continue
-        scheme = "https" if config.yarn_rm_url and config.yarn_rm_url.startswith("https://") else "http"
-        nm_url = f"{scheme}://{node_http_address}/node/containerlogs/{container_id}/{user}/"
-        resp = _request("GET", nm_url, timeout=60.0, verify=config.verify_for_httpx(), auth=config.auth_for_httpx())
-        if resp.status_code >= 400:
-            raise _logs_unavailable_error(application_id)
-        parts.append(f"=== container: {container_id} ===\n{resp.text}")
-
-    if not parts:
-        raise _logs_unavailable_error(application_id)
-    return "\n\n".join(parts)
+    return resp.text
 
 
 def get_application_logs(application_id: str, config: YarnClientConfig) -> str:
@@ -213,7 +196,7 @@ def get_application_logs(application_id: str, config: YarnClientConfig) -> str:
     url = f"{_base_url(config.yarn_rm_url)}/ws/v1/cluster/apps/{application_id}/aggregated-logs"
     resp = _request("GET", url, timeout=60.0, verify=config.verify_for_httpx(), auth=config.auth_for_httpx())
     if resp.status_code in (404, 501):
-        return _fetch_logs_via_nodemanager(application_id, config)
+        return _fetch_logs_via_am_container(application_id, config)
     if resp.status_code >= 400:
         logger.error(f"YARN GET {url} -> {resp.status_code}: {resp.text[:500]}")
         raise YarnError(f"YARN GET logs returned HTTP {resp.status_code}")
diff --git a/tests/unit/test_yarn_client.py b/tests/unit/test_yarn_client.py
index 7a3024f..6379e42 100644
--- a/tests/unit/test_yarn_client.py
+++ b/tests/unit/test_yarn_client.py
@@ -125,52 +125,40 @@ def test_logs_returns_text_on_h3_aggregated_endpoint():
     assert m.call_args.kwargs["verify"] is True
 
 
-def test_logs_falls_back_to_nodemanager_on_404():
+def test_logs_falls_back_to_am_container_on_404():
     responses = [
-        _resp(404),  # aggregated-logs H3 endpoint missing
-        _resp(200, json_data={"appAttempts": {"appAttempt": [{"id": "attempt_1"}]}}),
+        _resp(404),  # aggregated-logs endpoint missing
         _resp(
             200,
             json_data={
-                "containers": {
-                    "container": [
-                        {
-                            "id": "container_1",
-                            "user": "hdfs",
-                            "nodeHttpAddress": "nm1:8042",
-                        }
-                    ]
+                "app": {
+                    "amContainerLogs": "http://nm1:8042/node/containerlogs/container_1/hdfs",
                 }
             },
         ),
-        _resp(200, text="container log content"),
+        _resp(200, text="am container log content"),
     ]
     with patch(
         "spark_executor.core.yarn_client.httpx.request", side_effect=responses
     ) as m:
         out = get_application_logs("application_1", YarnClientConfig(yarn_rm_url=RM))
-    assert "container_1" in out
-    assert "container log content" in out
+    assert out == "am container log content"
     calls = m.call_args_list
     assert calls[0].args == (
         "GET",
         f"{RM}/ws/v1/cluster/apps/application_1/aggregated-logs",
     )
-    assert calls[1].args == ("GET", f"{RM}/ws/v1/cluster/apps/application_1/appattempts")
+    assert calls[1].args == ("GET", f"{RM}/ws/v1/cluster/apps/application_1")
     assert calls[2].args == (
         "GET",
-        f"{RM}/ws/v1/cluster/apps/application_1/appattempts/attempt_1/containers",
-    )
-    assert calls[3].args == (
-        "GET",
-        "http://nm1:8042/node/containerlogs/container_1/hdfs/",
+        "http://nm1:8042/node/containerlogs/container_1/hdfs/stdout",
     )
 
 
-def test_logs_raises_on_404_when_nm_also_empty():
+def test_logs_raises_on_404_when_am_container_field_missing():
     responses = [
-        _resp(404),  # aggregated-logs H3 endpoint missing
-        _resp(200, json_data={"appAttempts": {"appAttempt": []}}),
+        _resp(404),  # aggregated-logs endpoint missing
+        _resp(200, json_data={"app": {}}),
     ]
     with patch(
         "spark_executor.core.yarn_client.httpx.request", side_effect=responses
@@ -179,47 +167,27 @@ def test_logs_raises_on_404_when_nm_also_empty():
             get_application_logs("application_1", YarnClientConfig(yarn_rm_url=RM))
 
 
-def test_logs_handles_multiple_containers():
+def test_logs_returns_am_container_stdout():
     responses = [
-        _resp(404),  # aggregated-logs H3 endpoint missing
-        _resp(200, json_data={"appAttempts": {"appAttempt": [{"id": "attempt_1"}]}}),
+        _resp(404),  # aggregated-logs endpoint missing
         _resp(
             200,
             json_data={
-                "containers": {
-                    "container": [
-                        {
-                            "id": "container_1",
-                            "user": "hdfs",
-                            "nodeHttpAddress": "nm1:8042",
-                        },
-                        {
-                            "id": "container_2",
-                            "user": "hdfs",
-                            "nodeHttpAddress": "nm2:8042",
-                        },
-                    ]
+                "app": {
+                    "amContainerLogs": "http://nm1:8042/node/containerlogs/container_1/hdfs",
                 }
             },
         ),
-        _resp(200, text="log one"),
-        _resp(200, text="log two"),
+        _resp(200, text="driver stdout"),
     ]
     with patch(
         "spark_executor.core.yarn_client.httpx.request", side_effect=responses
     ) as m:
         out = get_application_logs("application_1", YarnClientConfig(yarn_rm_url=RM))
-    assert "log one" in out
-    assert "log two" in out
-    assert out.index("log one") < out.index("log two")
-    calls = m.call_args_list
-    assert calls[3].args == (
+    assert out == "driver stdout"
+    assert m.call_args_list[2].args == (
         "GET",
-        "http://nm1:8042/node/containerlogs/container_1/hdfs/",
-    )
-    assert calls[4].args == (
-        "GET",
-        "http://nm2:8042/node/containerlogs/container_2/hdfs/",
+        "http://nm1:8042/node/containerlogs/container_1/hdfs/stdout",
     )