fix(cpg): stop a failed session start from consuming the transaction budget

The seven-class profile showed a 25.3 s head "build" of which the trace attributes
only 4.5 s to the frontend. The remaining twenty seconds were the session's startup
poll, which was bounded by `min(startup_timeout, deadline - now)` and therefore ran
until the transaction deadline was exhausted. The fallback then had nothing left,
so the transport's failure mode consumed exactly the budget it existed to save.

A session now refuses to start unless enough budget survives for the work it exists
to serve, and the wait is bounded by its own allowance rather than by the whole
deadline. Under the 30-second hook deadline it declines immediately instead of
burning what is left.

Controls cover both: refusing is immediate and starts no process, and a startup
wait ends at its allowance nowhere near a long deadline.

This does not make the gate reachable; it removes a self-inflicted cost that was
hiding the real one. 1,401 tests passed, nine skipped.

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

src/openultrasast/cpg/session.py

@@ -71,6 +71,13 @@ class EngineSession:
     env: dict[str, str]
     joern: str = "joern"
     startup_timeout_seconds: float = 180.0
+    # Startup measured 9 to 13 seconds. Waiting for it out of the transaction's own budget is only
+    # worth doing when enough budget survives to run the work the session exists to serve, and the
+    # wait itself must be bounded by an allowance rather than by the whole deadline. Without both,
+    # a session that cannot start in time consumes every second that was left and the fallback runs
+    # with nothing: measured as a 25.3 s "build" of which only 4.5 s was the frontend.
+    startup_allowance_seconds: float = 20.0
+    work_reserve_seconds: float = 10.0
     failure: str = ""
     startup_seconds: float = 0.0
     loaded_graph: str = ""
@@ -113,6 +120,11 @@ class EngineSession:
         if self._remaining(self.startup_timeout_seconds) <= 0:
             self._poison("deadline_exhausted")
             return False
+        available = self.budget.deadline_monotonic - time.monotonic()
+        if available < self.startup_allowance_seconds + self.work_reserve_seconds:
+            # Refuse rather than spend the caller's remaining budget discovering this.
+            self._poison("session_budget_insufficient")
+            return False
         self.scratch.mkdir(parents=True, exist_ok=True)
         self._port = _free_port()
         started = time.monotonic()
@@ -134,7 +146,8 @@ class EngineSession:
             if self._process.poll() is not None:
                 self._poison("session_exited_during_startup")
                 return False
-            remaining = self._remaining(self.startup_timeout_seconds - (time.monotonic() - started))
+            allowance = min(self.startup_timeout_seconds, self.startup_allowance_seconds)
+            remaining = self._remaining(allowance - (time.monotonic() - started))
             if remaining <= 0:
                 self._poison("session_startup_timeout")
                 return False

tests/test_cpg_session.py

@@ -67,7 +67,8 @@ def fake(tmp_path):
     return script
 
 
-def session(tmp_path, fake, *, mode="ok", seconds=30.0, allowance=2.0):
+def session(tmp_path, fake, *, mode="ok", seconds=120.0, allowance=2.0):
+    """A budget comfortably above the session's startup allowance plus work reserve."""
     return EngineSession(
         tmp_path / "scratch",
         ExecutionBudget(time.monotonic() + seconds, allowance),
@@ -120,7 +121,8 @@ def test_a_crashed_session_answers_nothing_afterwards(tmp_path, fake):
 
 
 def test_cancellation_kills_the_process_group(tmp_path, fake):
-    live = session(tmp_path, fake, mode="hang", seconds=3.0)
+    live = session(tmp_path, fake, mode="hang", seconds=40.0)
+    live.work_reserve_seconds = 1.0
     assert live.start()
     pid = live._process.pid
     started = time.monotonic()
@@ -160,7 +162,7 @@ def test_the_wrapper_writes_its_receipt_before_running_the_body():
 
 
 def test_a_launch_failure_is_recorded_not_raised(tmp_path):
-    live = EngineSession(tmp_path / "s", ExecutionBudget(time.monotonic() + 10, 1.0), dict(os.environ), joern=str(tmp_path / "absent"))
+    live = EngineSession(tmp_path / "s", ExecutionBudget(time.monotonic() + 120, 1.0), dict(os.environ), joern=str(tmp_path / "absent"))
     assert live.start() is False and live.failure.startswith("session_launch_failed")
     live.close()
 
@@ -317,3 +319,27 @@ def test_the_driver_ends_the_session_with_the_transaction(tmp_path):
     (tmp_path / "a.js").write_text("const x = 1;\n")
     scan_repository(tmp_path, [], backend=backend)
     assert closed == [True]
+
+
+def test_a_session_refuses_to_start_when_the_budget_cannot_afford_it(tmp_path, fake):
+    """M2 11.6: a failed start must not consume the budget the fallback needs."""
+    live = session(tmp_path, fake, seconds=25.0)  # below allowance plus reserve
+    started = time.monotonic()
+    assert live.start() is False
+    assert live.failure == "session_budget_insufficient"
+    assert time.monotonic() - started < 1.0, "refusing must be immediate, not a wait"
+    assert live._process is None
+    live.close()
+
+
+def test_a_startup_wait_is_bounded_by_its_allowance_not_the_deadline(tmp_path, fake, monkeypatch):
+    live = session(tmp_path, fake, seconds=600.0)
+    live.startup_allowance_seconds = 1.0
+    live.work_reserve_seconds = 1.0
+    monkeypatch.setattr(EngineSession, "_post", lambda self, code, timeout: None)
+    started = time.monotonic()
+    assert live.start() is False
+    assert live.failure == "session_startup_timeout"
+    # Bounded by the allowance, nowhere near the 600-second deadline.
+    assert time.monotonic() - started < 10.0
+    live.close()