//// Partially executed cells must not replay side effects without explicit permission.

import albedo/harness/extensions/python/cells
import albedo/harness/extensions/python/kernel as python
import albedo/harness/runtime
import gleam/string
import gleeunit/should

pub fn partial_execution_requires_explicit_replay_permission_test() -> Nil {
  let assert Ok(host) = runtime.start(":memory:")
  let assert Ok(session) = runtime.open_session(host, "a", "/tmp")
  let assert Ok(failed) =
    runtime.execute(
      host,
      session,
      "counter = globals().get('counter', 0) + 1\nraise ValueError('bad')",
      5000,
    )
  let assert Ok(original) = runtime.cell(host, failed.cell_id)
  original.started |> should.be_true
  let repair =
    "await cells.run('"
    <> failed.cell_id
    <> "', replacements=[(\"raise ValueError('bad')\", 'counter')]"
  let assert Ok(denied) = runtime.execute(host, session, repair <> ")", 5000)
  let assert Ok(denied) = denied.result
  denied.status |> should.equal(python.Failed)
  denied.output |> string.contains("allow_partial=True") |> should.be_true
  let assert Ok(counter) = runtime.execute(host, session, "counter", 5000)
  let assert Ok(counter) = counter.result
  counter.value |> should.equal("1")
  let assert Ok(allowed) =
    runtime.execute(host, session, repair <> ", allow_partial=True)", 5000)
  let assert Ok(allowed) = allowed.result
  allowed.value |> should.equal("2")
  runtime.stop(host)
}

pub fn ambiguous_repairs_and_compile_errors_do_not_execute_test() -> Nil {
  let assert Ok(host) = runtime.start(":memory:")
  let assert Ok(session) = runtime.open_session(host, "a", "/tmp")
  let assert Ok(failed) =
    runtime.execute(host, session, "state = 1\n(yield 1)", 5000)
  let assert Ok(original) = runtime.cell(host, failed.cell_id)
  original.started |> should.be_false
  let assert Ok(denied) =
    runtime.execute(
      host,
      session,
      "await cells.run('" <> original.id <> "', replacements=[('1', '2')])",
      5000,
    )
  let assert Ok(denied) = denied.result
  denied.status |> should.equal(python.Failed)
  denied.output |> string.contains("exactly once") |> should.be_true
  let assert Ok(state) =
    runtime.execute(host, session, "'state' in globals()", 5000)
  let assert Ok(state) = state.result
  state.value |> should.equal("False")
  runtime.stop(host)
}

pub fn reused_call_id_does_not_fail_unique_constraint_test() -> Nil {
  let assert Ok(host) = runtime.start(":memory:")
  let storage = runtime.ledger(host)
  let session = "test-session"
  let assert Ok(first_id) =
    cells.begin_call(storage, session, "call_0", "x = 1")
  first_id |> should.equal(session <> "/call_0")

  let assert Ok(second_id) =
    cells.begin_call(storage, session, "call_0", "x = 2")
  string.starts_with(second_id, session <> "/call_0-") |> should.be_true
  { first_id != second_id } |> should.be_true

  let assert Ok(found) = cells.find_call(storage, session <> "/call_0")
  found.id |> should.equal(second_id)

  runtime.stop(host)
}