fix(resource_plan): take last two lscpu fields, not [1]/[2] (#332 review)

JustVugg caught a regression by running the PR: the code asked
`lscpu -p=CPU,Core,Socket` and read fields[1]/fields[2], but the comment's
claim was inverted -- `lscpu -p=<list>` emits EXACTLY the requested columns
(no CPU prefix), while bare `lscpu -p` prepends CPU.

On machines where the requested list is short or lscpu collapses to two
columns, every line was skipped by the `< 3` guard, cores stayed empty, and
the code fell through to os.cpu_count() -- the LOGICAL count. Result: 6
physical cores reported as 12 (SMT over-subscription), the opposite of the
fix.

Correct per review:
  - ask lscpu for exactly 'core,socket'
  - take fields[-2], fields[-1] (correct whether or not CPU is prepended)
  - keep the warning scaffolding + the _resolve_physical_cores clamp
    (those are the actual #325 fix; only the indexing was wrong)

Test now exercises BOTH the 2-column (-p=core,socket) and 3-column
(bare -p, CPU prefix) layouts, asserting 12 physical cores each. The
2-column case is the one that regressed: old parser returns 24, new
returns 12.

c/resource_plan.py

@@ -212,24 +212,31 @@ def physical_cpu_count():
         except (OSError, ValueError, AttributeError) as error:
             _physical_cores_warn(f"Windows core probe failed: {error}")
     try:
-        # lscpu -p prepende SEMPRE la colonna CPU, quindi `-p=core,socket` emette
-        # in realta' "CPU,Core,Socket" (3 colonne). Dobbiamo leggere esplicitamente
-        # CPU/Core/Socket e deduplicare su (core, socket): contare le righe non
-        # deduplicate restituirebbe i thread logici (SMT) - l'errore originale.
-        # I campi vuoti ("-") marcano core/socket offline e non sono interi.
-        result = subprocess.run(["lscpu", "-p=CPU,Core,Socket"], text=True,
+        # Ask lscpu for exactly core,socket and dedupe on (core, socket).
+        # Counting un-deduplicated rows would return logical threads (SMT),
+        # which was the original over-subscription bug. Empty fields ("-")
+        # mark an offline core/socket and fail int() -> skipped.
+        #
+        # Column layout robustness: `lscpu -p=<list>` emits *exactly* the
+        # requested columns (no CPU prefix), while bare `lscpu -p` prepends
+        # CPU. We requested two columns, but take the LAST TWO fields so the
+        # parser stays correct whether or not a CPU column is present
+        # (JustVugg review: the previous fields[1]/fields[2] indexing assumed
+        #  a 3-column layout and regressed 2-column output to the logical
+        # count -- the opposite of the fix).
+        result = subprocess.run(["lscpu", "-p=core,socket"], text=True,
                                 capture_output=True, check=True, timeout=5)
         cores = set()
         for line in result.stdout.splitlines():
             if not line or line.startswith("#"):
                 continue
             fields = line.split(",")
-            if len(fields) < 3:
+            if len(fields) < 2:
                 continue
             try:
-                core, socket = int(fields[1]), int(fields[2])
+                core, socket = int(fields[-2]), int(fields[-1])
             except ValueError:
-                continue  # "-" per un core/socket offline
+                continue  # "-" for an offline core/socket
             cores.add((core, socket))
         if cores:
             return len(cores)

c/tests/test_resource_plan.py

@@ -98,16 +98,31 @@ class ResourcePlanTest(unittest.TestCase):
             return subprocess.CompletedProcess(args=[], returncode=0,
                                                stdout=stdout, stderr="")
         # 1 socket, 12 cores, 2 SMT siblings -> 24 threads, 12 physical cores.
-        # lscpu prepends the CPU column, so output is CPU,Core,Socket.
-        rows = [f"{cpu},{core},0" for core in range(12) for cpu in range(2)]
-        blob = "# CPU,Core,Socket\n" + "\n".join(rows)
-        with mock.patch("resource_plan.subprocess.run", return_value=lscpu(blob)), \
-             mock.patch.object(sys, "platform", "linux"):
-            plan = build_plan(self.model, available_memory=16 * GB,
-                              available_disk=1, gpus=[])
-            env = environment_for_plan(plan)
-        self.assertEqual(plan["cpu"]["physical_cores"], 12)
-        self.assertEqual(env["OMP_NUM_THREADS"], "12")
+
+        # The parser must return 12 physical cores under BOTH lscpu layouts:
+        #  - 2-col: `lscpu -p=core,socket` emits exactly [core,socket] (this is
+        #           what the probe actually requests; the previous fields[1]/[2]
+        #           indexing skipped every line here and fell through to the
+        #           logical count -> the regression JustVugg caught).
+        #  - 3-col: bare `lscpu -p` prepends a CPU column -> [cpu,core,socket].
+        # Taking the last two fields is correct in both cases.
+        layouts = {
+            "2-col (-p=core,socket)": (
+                "# core,socket\n" +
+                "\n".join(f"{core},0" for core in range(12) for _ in range(2))),
+            "3-col (bare -p, CPU prefix)": (
+                "# CPU,Core,Socket\n" +
+                "\n".join(f"{cpu},{core},0" for core in range(12) for cpu in range(2))),
+        }
+        for label, blob in layouts.items():
+            with mock.patch("resource_plan.subprocess.run",
+                            return_value=lscpu(blob)), \
+                 mock.patch.object(sys, "platform", "linux"):
+                plan = build_plan(self.model, available_memory=16 * GB,
+                                  available_disk=1, gpus=[])
+                env = environment_for_plan(plan)
+            self.assertEqual(plan["cpu"]["physical_cores"], 12, label)
+            self.assertEqual(env["OMP_NUM_THREADS"], "12", label)
 
     def test_plan_does_not_set_omp_affinity_vars(self):
         # The real #325 regression: --auto-tier set OMP_PROC_BIND=spread +