//// Process-group termination must clean up descendants and races beyond normal E2E teardown.
//// Kernel-level supervision: owned process groups, quota accounting, verified stops.

import albedo/daemon/store
import albedo/harness/extensions/python/kernel as python
import gleam/int
import gleeunit/should
import harness/kernel_fixture

/// A group and the descendant that outlived the command must become
/// absent. Permission errors do not prove termination.
const reaped_probe =
  "import os, time\nstate = 'alive'\nfor _ in range(50):\n"
  <> "    reachable = 0\n"
  <> "    for probe in (lambda: os.killpg(job.group.pgid, 0), lambda: os.kill(grandchild, 0)):\n"
  <> "        try:\n            probe()\n            reachable += 1\n"
  <> "        except ProcessLookupError:\n            pass\n"
  <> "    if reachable == 0:\n        state = 'gone'\n        break\n"
  <> "    time.sleep(0.02)\n"

/// Parent programs for a job, as `program`: each prints the pid of a sleeping
/// child that stays in the job's group, then waits for it or leaves it behind.
const waits_on_a_sleeper =
  "program = 'import subprocess; child = subprocess.Popen([\"sleep\", \"30\"]); print(child.pid, flush=True); child.wait()'\n"

const leaves_a_sleeper =
  "program = 'import subprocess; child = subprocess.Popen([\"sleep\", \"30\"]); print(child.pid, flush=True)'\n"

pub fn deadline_ends_the_group_and_reports_it_test() -> Nil {
  let assert Ok(store) = store.start(":memory:", "")
  let assert Ok(kernel) = kernel_fixture.local(store, "/tmp")
  let assert Ok(outcome) =
    python.execute(
      kernel,
      "a",
      waits_on_a_sleeper
        // The deadline must outlast the parent's startup, or the job ends
        // before the grandchild it is meant to take down exists.
        <> "import os, sys\njob = run(sys.executable, '-c', program, timeout=1)\nawait job\n"
        <> "grandchild = int(job.tail().split()[0])\n"
        <> reaped_probe
        <> "(job.timed_out, job.termination.gone, state, job.poll() < 0, "
        <> "'deadline exceeded after 1s' in job.tail(), 'terminated' in job.tail())",
      10_000,
    )
  outcome.status |> should.equal(python.Succeeded)
  outcome.value |> should.equal("(True, True, 'gone', True, True, True)")
  let assert Ok(_) = python.stop(kernel)
  store.close(store)
}

pub fn command_exit_ends_a_descendant_that_holds_output_test() -> Nil {
  let assert Ok(store) = store.start(":memory:", "")
  let assert Ok(kernel) = kernel_fixture.local(store, "/tmp")
  let assert Ok(outcome) =
    python.execute(
      kernel,
      "a",
      leaves_a_sleeper
        <> "import os, sys, time\nstarted = time.monotonic()\n"
        <> "job = run(sys.executable, '-c', program, timeout=30)\nawait job\n"
        <> "grandchild = int(job.tail().split()[0])\n"
        <> reaped_probe
        <> "(job.timed_out, job.termination.gone, state, time.monotonic() - started < 5)",
      10_000,
    )
  outcome.status |> should.equal(python.Succeeded)
  outcome.value |> should.equal("(False, True, 'gone', True)")
  let assert Ok(_) = python.stop(kernel)
  store.close(store)
}

pub fn stop_ends_a_running_job_group_test() -> Nil {
  let assert Ok(store) = store.start(":memory:", "")
  let assert Ok(kernel) = kernel_fixture.local(store, "/tmp")
  let assert Ok(started) =
    python.execute(
      kernel,
      "a",
      "import asyncio\nlate = run('sleep', '30', timeout=300)\nawait asyncio.sleep(0.3)\nlate.group.pgid",
      10_000,
    )
  let assert Ok(pgid) = int.parse(started.value)
  // The typed stop is the supervisor's verdict: every owned group is gone.
  let assert Ok(_) = python.stop(kernel)
  let assert Ok(probe) = kernel_fixture.local(store, "/tmp")
  let assert Ok(checked) =
    python.execute(
      probe,
      "b",
      "import os\nstate = 'alive'\ntry:\n    os.killpg("
        <> int.to_string(pgid)
        <> ", 0)\nexcept ProcessLookupError:\n    state = 'gone'\nstate",
      5000,
    )
  checked.value |> should.equal("'gone'")
  let assert Ok(_) = python.stop(probe)
  store.close(store)
}