// A complete, bounded coroutine experiment using standalone Asio 1.38.2.
// One io_context runner, one operation at a time, no detached work or sleeps.
#include <asio/awaitable.hpp>
#include <asio/bind_cancellation_slot.hpp>
#include <asio/cancellation_signal.hpp>
#include <asio/co_spawn.hpp>
#include <asio/io_context.hpp>
#include <asio/post.hpp>
#include <asio/redirect_error.hpp>
#include <asio/steady_timer.hpp>
#include <asio/this_coro.hpp>
#include <asio/use_awaitable.hpp>
#include <array>
#include <chrono>
#include <exception>
#include <iostream>
#include <memory>
#include <stdexcept>
#include <string>
#include <utility>

namespace {
enum class Mode { succeed, cancel, fail };
enum class Event { started, other_handler, resumed, completed };
struct Control {
  asio::cancellation_signal cancel;
  std::array<Event,4> events{};
  std::size_t count{};
  void record(Event event) {
    if (count==events.size()) throw std::logic_error{"trace exceeded its bound"};
    events[count++]=event;
  }
};
struct Payload { std::string label; };
struct Outcome { bool cancelled{}; std::string label; };

// Named coroutine + value parameters: the frame owns these shared owners.
// The payload has no owner outside the suspended operation after caller.reset().
asio::awaitable<Outcome> wait_owned(std::shared_ptr<const Payload> payload,
                                    std::shared_ptr<Control> control, Mode mode) {
  const auto executor=co_await asio::this_coro::executor;
  asio::steady_timer timer{executor};
  timer.expires_at(mode==Mode::cancel ? std::chrono::steady_clock::time_point::max()
                                    : std::chrono::steady_clock::now());
  control->record(Event::started);
  if (mode==Mode::cancel) {
    // post never invokes this handler inline. There is only one run() thread:
    // the coroutine reaches its pending wait before this handler can execute.
    asio::post(executor,[control] {
      control->record(Event::other_handler);
      control->cancel.emit(asio::cancellation_type::terminal);
    });
  }
  asio::error_code error;
  co_await timer.async_wait(asio::redirect_error(asio::use_awaitable,error));
  control->record(Event::resumed);
  if (error==asio::error::operation_aborted) co_return Outcome{true,payload->label};
  if (error) throw asio::system_error{error};
  if (mode==Mode::fail) throw std::runtime_error{"injected preparation failure"};
  co_return Outcome{false,payload->label};
}
void check(bool condition,const char* message) {
  if (!condition) throw std::runtime_error{message};
}
void run_case(Mode mode) {
  asio::io_context context;
  auto control=std::make_shared<Control>();
  auto caller=std::make_shared<const Payload>(Payload{"owned study"});
  const std::weak_ptr<const Payload> probe=caller;
  std::exception_ptr failure;
  Outcome result;
  std::size_t completions{};
  // This is an ordinary completion handler, NOT a capturing coroutine lambda.
  // References stay valid because run() drains the operation before return.
  asio::co_spawn(context,wait_owned(caller,control,mode),
    asio::bind_cancellation_slot(control->cancel.slot(),
      [&](std::exception_ptr error,Outcome value) {
        ++completions;
        failure=error;
        result=std::move(value);
        control->record(Event::completed);
      }));
  caller.reset();
  check(!probe.expired(),"suspended operation must own its payload");
  context.run(); // Graceful drain: do not use stop() to skip cleanup completions.
  check(completions==1,"one completion must be observed");
  check(probe.expired(),"frame/payload must be released after completion");
  check(control->events[0]==Event::started,"coroutine entry recorded");
  check(control->events[control->count-1]==Event::completed,"completion recorded last");
  if (mode==Mode::fail) {
    check(static_cast<bool>(failure),"coroutine exception must reach completion");
    try { std::rethrow_exception(failure); }
    catch (const std::runtime_error& error) {
      check(std::string{error.what()}=="injected preparation failure","expected injected exception");
    }
    std::cout << "failure: exception observed; payload released\n";
  } else {
    check(!failure,"unexpected coroutine exception");
    check(result.label=="owned study","owned input survives suspension");
    check(result.cancelled==(mode==Mode::cancel),"expected terminal outcome");
    if (mode==Mode::cancel) {
      check(control->count==4 && control->events[1]==Event::other_handler &&
            control->events[2]==Event::resumed,"wait -> other handler -> cancelled resume -> completion");
      std::cout << "cancel: started -> other handler -> aborted wait -> completion; payload released\n";
    } else std::cout << "success: expired wait -> completion; payload released\n";
  }
}
}
int main() {
  try {
    run_case(Mode::succeed);
    run_case(Mode::cancel);
    run_case(Mode::fail);
    std::cout << "3 bounded coroutine scenarios passed\n";
    return 0;
  } catch (const std::exception& error) {
    std::cerr << "coroutine experiment: " << error.what() << '\n';
    return 1;
  }
}