perf(cpg): run the overlay pass in-session and verify it saved (M2 11.4)

The overlay was the last fixed cost bypassing the session: a whole JVM start,
15.8 s in the measured Node runs, while the census and query batches already route
through it.

`save` writes under the process working directory, which for a session is its own
private scratch rather than the build scratch, so the saved graph is now looked for
in both. Finding neither keeps the raw graph and records it rather than reporting
success, because a graph without its dataflow layer has that layer recomputed by
every later batch, 54 s each on a 637-file plugin, with nothing otherwise saying so.

The disposable invocation remains the fallback and runs unchanged whenever the
session is off or cannot answer.

Controls cover both new paths: an engine that reports success but saves no graph
keeps the raw one and notes it, and a graph saved under the session's workspace is
found and applied.

1,395 tests passed, nine skipped. The transport stays off by default; task 11.5
measures against the frozen profile. Rollout stays NO-GO.

Co-Authored-By: Claude Opus 5 (1M context) <[email protected]>
Claude-Session: https://claude.ai/code/session_01EmM1j46qgKMh3zqD1QMwEo

src/openultrasast/cpg/backend.py

@@ -923,24 +923,42 @@ class JoernBackend:
         A failure here is logged and the raw graph kept, so the scan degrades to the old cost rather than
         losing its graph.
         """
-        joern = shutil.which("joern")
-        if joern is None:
-            return False
-        script = self.queries_dir / "overlay.sc"
-        command = [joern, self._heap_flag(), "--script", str(script), "--param", f"cpgFile={cpg_path}"]
-        completed = self._run(command, timeout=self.build_timeout, cwd=scratch)
-        workspace = scratch / "workspace"
-        saved = workspace / cpg_path.name / "cpg.bin"
-        try:
+        # M2 11.4: the overlay is a whole JVM start, 15.8 s in the measured Node runs, and it is the
+        # last fixed cost that bypassed the session. `save` writes under the PROCESS working
+        # directory, which for a session is its own scratch and not this one, so the saved graph is
+        # looked for in both. Finding neither keeps the raw graph rather than reporting success:
+        # a graph without its dataflow layer has that layer recomputed by every later batch, 54 s
+        # each on a 637-file plugin, and nothing would otherwise say so.
+        session = self._session_for(cpg_path)
+        workspaces = [scratch / "workspace"]
+        answered = False
+        if session is not None:
+            workspaces.insert(0, session.scratch / "workspace")
+            body = self._session_payload(cpg_path, "overlay", {"cpgFile": str(cpg_path)})
+            answered = body is not None and extract_payload(body) is not None
+        if not answered:
+            joern = shutil.which("joern")
+            if joern is None:
+                return False
+            script = self.queries_dir / "overlay.sc"
+            command = [joern, self._heap_flag(), "--script", str(script), "--param", f"cpgFile={cpg_path}"]
+            completed = self._run(command, timeout=self.build_timeout, cwd=scratch)
             answered = completed is not None and completed.returncode == 0 and extract_payload(completed.stdout or "") is not None
-            if not answered or not saved.is_file():
+            if not answered:
                 detail = (completed.stderr or completed.stdout or "")[-400:] if completed is not None else "timeout"
                 logger.warning("overlays could not be applied to %s, keeping the raw graph: %s", cpg_path, detail)
+        saved = next((w / cpg_path.name / "cpg.bin" for w in workspaces if (w / cpg_path.name / "cpg.bin").is_file()), None)
+        try:
+            if not answered or saved is None:
+                if answered:
+                    self._note("overlay_saved_graph_missing")
+                    logger.warning("overlays reported success but saved no graph for %s, keeping the raw graph", cpg_path)
                 return False
             saved.replace(cpg_path)
             return True
         finally:
-            self._cleanup(workspace)
+            for workspace in workspaces:
+                self._cleanup(workspace)
 
     def _build_with_retries(
         self, root: Path, cpg_path: Path, scratch: Path, language: str, exclude: Sequence[str] = ()