///|
fn connection_require_open(
  state : ConnectionState,
) -> Unit raise ConnectionError {
  match state.phase {
    ConnectionOpen => ()
    ConnectionClosed(reason~) => raise Closed(reason~)
  }
}

///|
fn connection_outbound_index(
  pending : Array[PendingOutbound],
  id : RequestId,
) -> Int? {
  for index, value in pending.iter2() {
    if value.id == id {
      return Some(index)
    }
  }
  None
}

///|
fn connection_inbound_index(
  pending : Array[PendingInbound],
  id : RequestId,
) -> Int? {
  for index, value in pending.iter2() {
    if value.id == id {
      return Some(index)
    }
  }
  None
}

///|
fn connection_response_id(response : JsonRpcResponse) -> RequestId {
  match response {
    Success(success) => success.id
    Error(failure) =>
      match failure.id {
        String(value) => String(value)
        Number(value) => Number(value)
        Null => Null
      }
  }
}

///|
fn connection_response_id_json(id : RequestId) -> JsonRpcId {
  match id {
    String(value) => String(value)
    Number(value) => Number(value)
    Null => Null
  }
}

///|
fn connection_request_id_json(id : RequestId) -> Json {
  match id {
    String(value) => Json::string(value)
    Number(value) => Json::number(value.to_double(), repr=value.to_string())
    Null => Json::null()
  }
}

///|
fn connection_cancel_notification(
  id : RequestId,
) -> JsonRpcNotification raise ConnectionError {
  let fields : Map[String, Json] = Map([])
  fields["requestId"] = connection_request_id_json(id)
  JsonRpcNotification::new(
    method_name="$/cancel_request",
    params=Json::object(fields),
  ) catch {
    _ =>
      raise InvalidCancellation(
        reason="failed to construct cancellation notification",
      )
  }
}

///|
fn connection_cancel_id(params : Json?) -> RequestId raise ConnectionError {
  match params {
    Some(Object(fields)) => {
      if fields.length() != 1 || !fields.contains("requestId") {
        raise InvalidCancellation(
          reason="$/cancel_request.params must contain only requestId",
        )
      }
      let request_id = match fields.get("requestId") {
        Some(value) => value
        None =>
          raise InvalidCancellation(
            reason="$/cancel_request.params.requestId is required",
          )
      }
      let envelope : Map[String, Json] = Map([])
      envelope["jsonrpc"] = Json::string("2.0")
      envelope["id"] = request_id
      envelope["method"] = Json::string("$/cancel_request")
      let decoded = @jsonrpc.jsonrpc_decode_json(Json::object(envelope)) catch {
        _ =>
          raise InvalidCancellation(
            reason="$/cancel_request.params.requestId must be a string, number, or null",
          )
      }
      match decoded {
        Request(request) => request.id
        _ =>
          raise InvalidCancellation(
            reason="$/cancel_request.params.requestId must be a string, number, or null",
          )
      }
    }
    Some(_) =>
      raise InvalidCancellation(
        reason="$/cancel_request.params must be an object",
      )
    None =>
      raise InvalidCancellation(reason="$/cancel_request.params is required")
  }
}

///|
fn connection_reduce_outgoing_request(
  state : ConnectionState,
  request : JsonRpcRequest,
) -> ConnectionStep raise ConnectionError {
  connection_require_open(state)
  if connection_outbound_index(state.outbound, request.id) is Some(_) ||
    state.settled_outbound.contains(request.id) {
    raise DuplicateOutbound(id=request.id)
  }
  let outbound = Array::copy(state.outbound)
  outbound.push({
    id: request.id,
    method_name: request.method_name,
    cancel_requested: false,
  })
  {
    state: { ..state, outbound, },
    commands: [WriteMessage(JsonRpcMessage::request(request))],
  }
}

///|
fn connection_reduce_outgoing_notification(
  state : ConnectionState,
  notification : JsonRpcNotification,
) -> ConnectionStep raise ConnectionError {
  connection_require_open(state)
  {
    state,
    commands: [WriteMessage(JsonRpcMessage::notification(notification))],
  }
}

///|
fn connection_reduce_incoming_request(
  state : ConnectionState,
  request : JsonRpcRequest,
) -> ConnectionStep raise ConnectionError {
  connection_require_open(state)
  if connection_inbound_index(state.inbound, request.id) is Some(_) ||
    state.settled_inbound.contains(request.id) {
    raise DuplicateInbound(id=request.id)
  }
  let inbound = Array::copy(state.inbound)
  inbound.push({
    id: request.id,
    method_name: request.method_name,
    cancel_requested: false,
  })
  { state: { ..state, inbound, }, commands: [DispatchRequest(request)] }
}

///|
fn connection_reduce_incoming_notification(
  state : ConnectionState,
  notification : JsonRpcNotification,
) -> ConnectionStep raise ConnectionError {
  connection_require_open(state)
  if notification.method_name == "$/cancel_request" {
    let id = connection_cancel_id(notification.params)
    connection_reduce_wire_cancel_inbound(state, id)
  } else {
    { state, commands: [DispatchNotification(notification)] }
  }
}

///|
/// Apply one inbound wire `$/cancel_request` with best-effort semantics.
///
/// ACP v1 makes the cancellation best-effort: the receiver MAY cancel the
/// matching activity and the only MUST is that the original request
/// eventually receives its response; the contract is silent on ids that were
/// never seen or already settled. Real ACP clients routinely send cancels
/// that race with an already-delivered response, so a well-formed cancel of
/// a non-live id is a traced no-op — no state change, no response, no close —
/// and the connection keeps serving traffic:
/// - never-seen id -> `CancelUnknownRequest`;
/// - already-settled id (the settle-then-cancel race) -> `CancelAlreadySettled`;
/// - live pending already carrying a cancel mark (a repeated best-effort
///   cancel) -> `CancelAlreadyCancelled`.
///
/// A cancel for a genuinely live pending keeps the exact live semantics of
/// `connection_reduce_cancel_inbound` (typed cancel mark plus the native
/// `CancelInboundTask` intent). Malformed cancellation params still fail
/// fast in `connection_cancel_id`: an invalid frame is a protocol violation,
/// not a benign race. The local `CancelInbound` event is deliberately NOT
/// tolerated: a local caller cancels engine-owned state, so an unknown,
/// settled, or repeated id there remains a fail-fast programming error.
fn connection_reduce_wire_cancel_inbound(
  state : ConnectionState,
  id : RequestId,
) -> ConnectionStep raise ConnectionError {
  match connection_inbound_index(state.inbound, id) {
    Some(index) =>
      if state.inbound[index].cancel_requested {
        {
          state,
          commands: [TraceCancelIgnored(id~, reason=CancelAlreadyCancelled)],
        }
      } else {
        connection_reduce_cancel_inbound(state, id)
      }
    None =>
      if state.settled_inbound.contains(id) {
        {
          state,
          commands: [TraceCancelIgnored(id~, reason=CancelAlreadySettled)],
        }
      } else {
        {
          state,
          commands: [TraceCancelIgnored(id~, reason=CancelUnknownRequest)],
        }
      }
  }
}

///|
fn connection_reduce_incoming_response(
  state : ConnectionState,
  response : JsonRpcResponse,
) -> ConnectionStep raise ConnectionError {
  let id = connection_response_id(response)
  match connection_outbound_index(state.outbound, id) {
    Some(index) => {
      let outbound = Array::copy(state.outbound)
      let _ = outbound.remove(index)
      let settled_outbound = Array::copy(state.settled_outbound)
      settled_outbound.push(id)
      {
        state: { ..state, outbound, settled_outbound },
        commands: [CompleteOutbound(id~, response~)],
      }
    }
    None =>
      match state.phase {
        ConnectionClosed(_) => {
          if state.settled_outbound.contains(id) {
            raise LateResponse(id~)
          }
          raise UnknownResponse(id~)
        }
        ConnectionOpen => {
          if state.settled_outbound.contains(id) {
            raise DuplicateResponse(id~)
          }
          raise UnknownResponse(id~)
        }
      }
  }
}

///|
fn connection_reduce_cancel_outbound(
  state : ConnectionState,
  id : RequestId,
) -> ConnectionStep raise ConnectionError {
  connection_require_open(state)
  match connection_outbound_index(state.outbound, id) {
    None => {
      if state.settled_outbound.contains(id) {
        raise DuplicateCancellation(id~)
      }
      raise UnknownCancellation(id~)
    }
    Some(index) => {
      let outbound = Array::copy(state.outbound)
      let _ = outbound.remove(index)
      let settled_outbound = Array::copy(state.settled_outbound)
      settled_outbound.push(id)
      let notification = connection_cancel_notification(id)
      {
        state: { ..state, outbound, settled_outbound },
        commands: [
          WriteMessage(JsonRpcMessage::notification(notification)),
          CancelOutboundTask(id~),
        ],
      }
    }
  }
}

///|
fn connection_reduce_cancel_inbound(
  state : ConnectionState,
  id : RequestId,
) -> ConnectionStep raise ConnectionError {
  connection_require_open(state)
  match connection_inbound_index(state.inbound, id) {
    None => {
      if state.settled_inbound.contains(id) {
        raise DuplicateCancellation(id~)
      }
      raise UnknownCancellation(id~)
    }
    Some(index) => {
      let inbound = Array::copy(state.inbound)
      let pending = inbound[index]
      if pending.cancel_requested {
        raise DuplicateCancellation(id~)
      }
      inbound[index] = { ..pending, cancel_requested: true }
      { state: { ..state, inbound, }, commands: [CancelInboundTask(id~)] }
    }
  }
}

///|
fn connection_reduce_incoming_completed(
  state : ConnectionState,
  id : RequestId,
  result : Json,
) -> ConnectionStep raise ConnectionError {
  match connection_inbound_index(state.inbound, id) {
    None => {
      if state.settled_inbound.contains(id) {
        raise LateInboundCompletion(id~)
      }
      raise UnknownInboundCompletion(id~)
    }
    Some(index) => {
      let inbound = Array::copy(state.inbound)
      let pending = inbound.remove(index)
      let settled_inbound = Array::copy(state.settled_inbound)
      settled_inbound.push(id)
      let response = if pending.cancel_requested {
        JsonRpcResponse::error(
          id=connection_response_id_json(id),
          error=JsonRpcError::request_cancelled(),
        )
      } else {
        JsonRpcResponse::success(id~, result~)
      }
      {
        state: { ..state, inbound, settled_inbound },
        commands: [WriteMessage(JsonRpcMessage::response(response))],
      }
    }
  }
}

///|
fn connection_reduce_incoming_failed(
  state : ConnectionState,
  id : RequestId,
  error : JsonRpcError,
) -> ConnectionStep raise ConnectionError {
  match connection_inbound_index(state.inbound, id) {
    None => {
      if state.settled_inbound.contains(id) {
        raise LateInboundCompletion(id~)
      }
      raise UnknownInboundCompletion(id~)
    }
    Some(index) => {
      let inbound = Array::copy(state.inbound)
      let pending = inbound.remove(index)
      let settled_inbound = Array::copy(state.settled_inbound)
      settled_inbound.push(id)
      let response = if pending.cancel_requested {
        JsonRpcResponse::error(
          id=connection_response_id_json(id),
          error=JsonRpcError::request_cancelled(),
        )
      } else {
        JsonRpcResponse::error(id=connection_response_id_json(id), error~)
      }
      {
        state: { ..state, inbound, settled_inbound },
        commands: [WriteMessage(JsonRpcMessage::response(response))],
      }
    }
  }
}

///|
fn connection_reduce_close(
  state : ConnectionState,
  reason : String,
) -> ConnectionStep raise ConnectionError {
  match state.phase {
    ConnectionClosed(reason=previous_reason) =>
      raise AlreadyClosed(reason=previous_reason)
    ConnectionOpen => {
      let commands : Array[ConnectionCommand] = []
      let settled_outbound = Array::copy(state.settled_outbound)
      for pending in state.outbound {
        settled_outbound.push(pending.id)
        commands.push(FailOutbound(id=pending.id, reason~))
      }
      let settled_inbound = Array::copy(state.settled_inbound)
      for pending in state.inbound {
        settled_inbound.push(pending.id)
        commands.push(
          FailInbound(
            id=pending.id,
            reason~,
            cancelled=pending.cancel_requested,
          ),
        )
      }
      commands.push(Trace(message="connection closed"))
      {
        state: {
          phase: ConnectionClosed(reason~),
          outbound: [],
          inbound: [],
          settled_outbound,
          settled_inbound,
        },
        commands,
      }
    }
  }
}

///|
/// Apply one event without performing any transport or runtime effect.
#warnings("-unused_value")
fn connection_reduce(
  state : ConnectionState,
  event : ConnectionEvent,
) -> ConnectionStep raise ConnectionError {
  match event {
    OutgoingRequest(request) =>
      connection_reduce_outgoing_request(state, request)
    OutgoingNotification(notification) =>
      connection_reduce_outgoing_notification(state, notification)
    IncomingRequest(request) =>
      connection_reduce_incoming_request(state, request)
    IncomingNotification(notification) =>
      connection_reduce_incoming_notification(state, notification)
    IncomingResponse(response) =>
      connection_reduce_incoming_response(state, response)
    IncomingCompleted(id~, result~) =>
      connection_reduce_incoming_completed(state, id, result)
    IncomingFailed(id~, error~) =>
      connection_reduce_incoming_failed(state, id, error)
    CancelOutbound(id~) => connection_reduce_cancel_outbound(state, id)
    CancelInbound(id~) => connection_reduce_cancel_inbound(state, id)
    Close(reason~) => connection_reduce_close(state, reason)
    Eof => connection_reduce_close(state, "eof")
  }
}