Per-project test execution, and fork-death diagnostics that name a cause

bleep-bsp-tests/src/scala/bleep/analysis/BuildStateReducerTest.scala

@@ -20,6 +20,44 @@ class BuildStateReducerTest extends AnyFunSuite with Matchers {
   private def reduce(events: BuildEvent*): BuildState =
     events.foldLeft(BuildState.empty)(BuildStateReducer.reduce)
 
+  // ==========================================================================
+  // Task-time / parallelism accounting
+  // ==========================================================================
+
+  test("totalTaskTimeMs counts a suite's whole fork occupancy, not the sum of its test durations") {
+    // A slow-booting integration suite spends most of its wall time on setup — starting an app and
+    // its containers — before any test method runs. The suite occupied its fork for 27s
+    // (SuiteStarted..SuiteFinished); the two test methods report 5ms each. Task time must be the 27s,
+    // or parallelism reads ~1x under fan-out.
+    val state = reduce(
+      BuildEvent.SuiteStarted(cpn("proj"), sn("com.example.SlowBootIT"), ts),
+      BuildEvent.TestFinished(cpn("proj"), sn("com.example.SlowBootIT"), tn("a"), TestStatus.Passed, 5, None, None, ts + 26000, None),
+      BuildEvent.TestFinished(cpn("proj"), sn("com.example.SlowBootIT"), tn("b"), TestStatus.Passed, 5, None, None, ts + 26500, None),
+      BuildEvent.SuiteFinished(cpn("proj"), sn("com.example.SlowBootIT"), SuiteOutcome.Executed(2, 0, 0, 0), 27000, ts + 27000)
+    )
+    state.totalTaskTimeMs shouldBe 27000L
+  }
+
+  test("totalTaskTimeMs sums two overlapping suites so parallelism exceeds wall time") {
+    // Two suites, each occupying a fork for 10s, started together: 20s of task time in 10s of wall.
+    val state = reduce(
+      BuildEvent.SuiteStarted(cpn("a"), sn("A"), ts),
+      BuildEvent.SuiteStarted(cpn("b"), sn("B"), ts),
+      BuildEvent.SuiteFinished(cpn("a"), sn("A"), SuiteOutcome.Executed(1, 0, 0, 0), 10000, ts + 10000),
+      BuildEvent.SuiteFinished(cpn("b"), sn("B"), SuiteOutcome.Executed(1, 0, 0, 0), 10000, ts + 10000)
+    )
+    state.totalTaskTimeMs shouldBe 20000L
+  }
+
+  test("a suite whose end is reported by both SuiteFinished and SuiteError is counted once") {
+    val state = reduce(
+      BuildEvent.SuiteStarted(cpn("proj"), sn("S"), ts),
+      BuildEvent.SuiteFinished(cpn("proj"), sn("S"), SuiteOutcome.Executed(0, 1, 0, 0), 8000, ts + 8000),
+      BuildEvent.SuiteError(cpn("proj"), sn("S"), "exited 1", bleep.bsp.protocol.ProcessExit.ExitCode(1), 8000, ts + 8001)
+    )
+    state.totalTaskTimeMs shouldBe 8000L
+  }
+
   // ==========================================================================
   // SuiteFinished synthetic failure tests
   // ==========================================================================

bleep-bsp-tests/src/scala/bleep/analysis/BuildSummaryVerdictTest.scala

@@ -23,6 +23,40 @@ class BuildSummaryVerdictTest extends AnyFunSuite with Matchers {
     state.toSummary(durationMs = 0L, wasCancelled = false).toEither
   }
 
+  /** Fold in the authoritative per-run counts the server sends as its response payload, exactly as `bleep test` and the MCP tools do via
+    * [[bleep.history.TranscriptFormat]] — the path where `suitesTotal`/`suitesCompleted` are set from discovery rather than accumulated from streamed events.
+    */
+  private def verdictWithResult(events: List[E], trr: BleepBspProtocol.TestRunResult): Either[bleep.BleepException, Unit] = {
+    val trrEvent = BuildEvent.TestRunCompleted(
+      totalPassed = trr.totalPassed,
+      totalFailed = trr.totalFailed,
+      totalSkipped = trr.totalSkipped,
+      totalIgnored = trr.totalIgnored,
+      suitesTotal = trr.suitesTotal,
+      suitesCompleted = trr.suitesCompleted,
+      suitesFailed = trr.suitesFailed,
+      suitesCancelled = trr.suitesCancelled,
+      durationMs = trr.durationMs,
+      timestamp = 0L
+    )
+    val state = (events.flatMap(BuildEvent.fromProtocol) :+ trrEvent).foldLeft(BuildState.empty)(BuildStateReducer.reduce)
+    state.toSummary(durationMs = 0L, wasCancelled = false).toEither
+  }
+
+  private def runResult(passed: Int, suitesTotal: Int, suitesCompleted: Int): BleepBspProtocol.TestRunResult =
+    BleepBspProtocol.TestRunResult(
+      totalPassed = passed,
+      totalFailed = 0,
+      totalSkipped = 0,
+      totalIgnored = 0,
+      suitesTotal = suitesTotal,
+      suitesCompleted = suitesCompleted,
+      suitesFailed = 0,
+      suitesCancelled = 0,
+      durationMs = 0L,
+      historyId = None
+    )
+
   private def leftMessage(events: List[E]): String =
     verdict(events).left.map(_.getMessage) match {
       case Left(msg) => msg
@@ -209,6 +243,38 @@ class BuildSummaryVerdictTest extends AnyFunSuite with Matchers {
     ) shouldBe Right(())
   }
 
+  test("a run that completed fewer suites than it discovered is not a pass, even with zero failures") {
+    // The `bleep test dfmt/test` report: 64 passed, 0 failed, yet only 10 of 29 discovered suites ran. `failed: 0` made it look green; the 19 that never
+    // reported a result are the whole story. The verdict must fail so CI's "test every project" gate cannot mistake a fraction for full coverage.
+    val msg = verdictWithResult(Nil, runResult(passed = 64, suitesTotal = 29, suitesCompleted = 10)).left.map(_.getMessage) match {
+      case Left(m)   => m
+      case Right(()) => fail("a run that left 19 of 29 suites unaccounted must be judged a failure")
+    }
+    msg should include("did not finish")
+    msg should include("10 of 29")
+    msg should include("19")
+  }
+
+  test("the same shortfall from streamed SuiteStarted/SuiteFinished events (batch announces every suite up front) also fails") {
+    // Per-project batch mode announces a SuiteStarted for every suite when the batch begins, then a SuiteFinished as each completes. If the fork stops after a
+    // subset, `suitesTotal` (announced) exceeds `suitesCompleted` (finished) with no failure event anywhere — the verdict has to catch it from the counts alone.
+    verdict(
+      List(
+        E.SuiteStarted(proj("app"), SuiteName("A"), timestamp = 1L),
+        E.SuiteStarted(proj("app"), SuiteName("B"), timestamp = 1L),
+        E.SuiteStarted(proj("app"), SuiteName("C"), timestamp = 1L)
+      ) ++ passedTest("app").map {
+        case ts: E.TestFinished  => ts.copy(suite = SuiteName("A"))
+        case sf: E.SuiteFinished => sf.copy(suite = SuiteName("A"))
+        case other               => other
+      }
+    ).isLeft shouldBe true
+  }
+
+  test("a run where every discovered suite completed is Right") {
+    verdictWithResult(passedTest("app"), runResult(passed = 42, suitesTotal = 29, suitesCompleted = 29)) shouldBe Right(())
+  }
+
   test("a clean run is Right") {
     verdict(compileFinished("app", CompileStatus.Success, skippedBecause = None) +: passedTest("app")) shouldBe Right(())
   }

bleep-bsp-tests/src/scala/bleep/analysis/LinkDagIntegrationTest.scala

@@ -269,6 +269,7 @@ class LinkDagIntegrationTest extends AnyFunSuite with Matchers {
         },
         discover = (_, _, _) => sys.error("DiscoverTask should not appear in a link DAG"),
         test = (_, _, _) => sys.error("TestSuiteTask should not appear in a link DAG"),
+        testBatch = (_, _) => sys.error("TestBatchTask should not appear in a link DAG"),
         sourcegen = (_, _) => sys.error("SourcegenTask should not appear here"),
         annotationProcessor = (_, _) => sys.error("ResolveAnnotationProcessorsTask should not appear here"),
         symbolProcessor = (_, _) => sys.error("ResolveSymbolProcessorsTask should not appear here")
@@ -316,6 +317,7 @@ class LinkDagIntegrationTest extends AnyFunSuite with Matchers {
           ),
         discover = (_, _, _) => sys.error("DiscoverTask should not appear in a link DAG"),
         test = (_, _, _) => sys.error("TestSuiteTask should not appear in a link DAG"),
+        testBatch = (_, _) => sys.error("TestBatchTask should not appear in a link DAG"),
         sourcegen = (_, _) => sys.error("SourcegenTask should not appear here"),
         annotationProcessor = (_, _) => sys.error("ResolveAnnotationProcessorsTask should not appear here"),
         symbolProcessor = (_, _) => sys.error("ResolveSymbolProcessorsTask should not appear here")
@@ -361,8 +363,9 @@ class LinkDagIntegrationTest extends AnyFunSuite with Matchers {
         mayAdmitCompile = _ => IO.pure(true),
         compile = (_, _) => IO.pure(TaskResult.Success),
         link = (_, _) => IO.pure((TaskResult.Failure("Link error", List.empty), LinkResult.Failure("Link error", List.empty))),
-        discover = (_, _, _) => IO.pure((TaskResult.Success, TaskDag.DiscoveryResult(Nil, 0, isTestProject = false))),
+        discover = (_, _, _) => IO.pure((TaskResult.Success, TaskDag.DiscoveryResult(Nil, 0, isTestProject = false, suiteParallelism = None, batches = Nil))),
         test = (_, _, _) => sys.error("TestSuiteTask should not appear in this DAG"),
+        testBatch = (_, _) => sys.error("TestBatchTask should not appear in this DAG"),
         sourcegen = (_, _) => sys.error("SourcegenTask should not appear here"),
         annotationProcessor = (_, _) => sys.error("ResolveAnnotationProcessorsTask should not appear here"),
         symbolProcessor = (_, _) => sys.error("ResolveSymbolProcessorsTask should not appear here")

bleep-bsp-tests/src/scala/bleep/analysis/MaxConcurrentSuitesDagTest.scala

@@ -0,0 +1,134 @@
+package bleep.analysis
+
+import bleep.bsp.{Outcome, TaskDag}
+import bleep.bsp.TaskDag.{TaskId, _}
+import bleep.model.{CrossProjectName, ProjectName, SuiteName}
+import bleep.testing.FrameworkSelection
+import cats.effect.{IO, Ref}
+import cats.effect.std.Queue
+import cats.effect.unsafe.implicits.global
+import org.scalatest.funsuite.AnyFunSuite
+import org.scalatest.matchers.should.Matchers
+
+/** `maxConcurrentSuites` turns a project's discovered suites into round-robin chains of ordering-only `runAfter` edges. The properties that matter:
+  *
+  *   - at bound 1 the suites run strictly sequentially in alphabetical order — surefire's usual class order, which schema-bootstrapping test setups rely on
+  *   - a failing suite does NOT skip the rest of its chain: `runAfter` is ordering, not failure propagation
+  *   - without a bound no chains exist and every suite is schedulable at once
+  */
+class MaxConcurrentSuitesDagTest extends AnyFunSuite with Matchers {
+
+  private def testMachine(cpu: Int): bleep.MachineResources =
+    bleep.MachineResources.create(totalCpu = cpu, totalMemoryMb = 64 * 1024, logger = ryddig.TypedLogger.DevNull, longWaitWarnMs = 60000L)
+
+  private def projectName(name: String): CrossProjectName =
+    CrossProjectName(ProjectName(name), None)
+
+  private val selection = FrameworkSelection.JUnitPlatform("junit")
+
+  /** Runs a one-project test DAG whose discovery yields `suiteNames` (deliberately unsorted) under the given parallelism bound. Returns the interleaved
+    * start/finish event log and the final dag.
+    */
+  private def run(
+      project: CrossProjectName,
+      suiteNames: List[String],
+      parallelism: Option[Int],
+      failingSuites: Set[String]
+  ): (List[String], Dag) = {
+    val dag = TaskDag.buildTestDag(
+      Set(project),
+      BuildContext(
+        allProjectDeps = Map.empty,
+        platforms = Map(project -> LinkPlatform.Jvm),
+        sourcegen = SourcegenPlan.empty,
+        apPlan = AnnotationProcessorPlan.empty,
+        kspPlan = SymbolProcessorPlan.empty,
+        testProjects = Set(project)
+      )
+    )
+
+    (for {
+      log <- Ref.of[IO, List[String]](Nil)
+      executor = TaskDag.executor(
+        Handlers(
+          mayAdmitCompile = _ => IO.pure(true),
+          compile = (_, _) => IO.pure(TaskResult.Success),
+          link = (_, _) => sys.error("no link on JVM"),
+          discover = (_, _, _) =>
+            IO.pure(
+              (
+                TaskResult.Success,
+                TaskDag.DiscoveryResult(
+                  suiteNames.map(_ -> selection),
+                  suiteNames.size,
+                  isTestProject = true,
+                  suiteParallelism = parallelism,
+                  batches = Nil
+                )
+              )
+            ),
+          test = (task, _, _) =>
+            for {
+              _ <- log.update(_ :+ s"start:${task.suiteName.value}")
+              // yield so that, were another suite schedulable, it could interleave between our start and finish
+              _ <- IO.cede
+              _ <- log.update(_ :+ s"finish:${task.suiteName.value}")
+            } yield if (failingSuites(task.suiteName.value)) TaskResult.Failure(s"${task.suiteName.value} failed", Nil) else TaskResult.Success,
+          testBatch = (_, _) => sys.error("no test batch here"),
+          sourcegen = (_, _) => sys.error("no sourcegen here"),
+          annotationProcessor = (_, _) => sys.error("no annotation processors here"),
+          symbolProcessor = (_, _) => sys.error("no symbol processors here")
+        )
+      )
+      eventQueue <- Queue.unbounded[IO, Option[DagEvent]]
+      killSignal <- Outcome.neverKillSignal
+      finalDag <- executor.execute(dag, testMachine(4), TaskDag.ForkHeaps.default, eventQueue, killSignal)
+      events <- log.get
+    } yield (events, finalDag)).unsafeRunSync()
+  }
+
+  test("suiteParallelism=1: suites run sequentially in alphabetical order") {
+    val project = projectName("app-test")
+    val (events, dag) = run(project, List("b.Suite", "c.Suite", "a.Suite"), parallelism = Some(1), failingSuites = Set.empty)
+
+    events shouldBe List("start:a.Suite", "finish:a.Suite", "start:b.Suite", "finish:b.Suite", "start:c.Suite", "finish:c.Suite")
+    dag.completed should contain allOf (
+      TaskId.Test(project, SuiteName("a.Suite")),
+      TaskId.Test(project, SuiteName("b.Suite")),
+      TaskId.Test(
+        project,
+        SuiteName("c.Suite")
+      )
+    )
+  }
+
+  test("suiteParallelism=1: a failing suite does not skip the rest of the chain") {
+    val project = projectName("app-test")
+    val (events, dag) = run(project, List("a.Suite", "b.Suite", "c.Suite"), parallelism = Some(1), failingSuites = Set("a.Suite"))
+
+    events shouldBe List("start:a.Suite", "finish:a.Suite", "start:b.Suite", "finish:b.Suite", "start:c.Suite", "finish:c.Suite")
+    dag.failed should contain(TaskId.Test(project, SuiteName("a.Suite")))
+    dag.skipped shouldBe empty
+    dag.completed should contain allOf (TaskId.Test(project, SuiteName("b.Suite")), TaskId.Test(project, SuiteName("c.Suite")))
+  }
+
+  test("suiteParallelism=2: each suite waits only for the one two places ahead of it") {
+    val project = projectName("app-test")
+    val (events, dag) = run(project, List("a.Suite", "b.Suite", "c.Suite"), parallelism = Some(2), failingSuites = Set.empty)
+
+    // c chains after a (index 2 waits for index 2-2=0); b is unchained
+    events.indexOf("start:c.Suite") should be > events.indexOf("finish:a.Suite")
+    dag.completed.count { case TaskId.Test(_, _) => true; case _ => false } shouldBe 3
+  }
+
+  test("no suiteParallelism: no ordering edges, all suites complete") {
+    val project = projectName("app-test")
+    val (events, dag) = run(project, List("b.Suite", "a.Suite"), parallelism = None, failingSuites = Set.empty)
+
+    events should have size 4
+    dag.completed should contain allOf (TaskId.Test(project, SuiteName("a.Suite")), TaskId.Test(project, SuiteName("b.Suite")))
+    dag.tasks.values.collect { case t: TestSuiteTask => t }.foreach { t =>
+      t.runAfter shouldBe empty
+    }
+  }
+}